Samir Boulahtit
f20266167d
- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.) - Added ignore rules for patterns intentional in this codebase: E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from), SIM108/SIM105/SIM117 (readability preferences) - Added per-file ignores for tests and scripts - Excluded broken scripts/rename_terminology.py (has curly quotes) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
273 lines
8.0 KiB
Python
273 lines
8.0 KiB
Python
# app/modules/config.py
|
|
"""
|
|
Module configuration auto-discovery utility.
|
|
|
|
Provides utilities for discovering and loading module-specific configurations.
|
|
Each self-contained module can have its own config.py file that defines:
|
|
- A Pydantic Settings class for environment-based configuration
|
|
- Default values for module settings
|
|
- Validation rules for configuration
|
|
|
|
Module Config Structure:
|
|
app/modules/<code>/
|
|
└── config.py # ModuleConfig class
|
|
|
|
Config File Pattern:
|
|
# app/modules/mymodule/config.py
|
|
from pydantic import Field
|
|
from pydantic_settings import BaseSettings
|
|
|
|
class MyModuleConfig(BaseSettings):
|
|
'''Configuration for mymodule.'''
|
|
|
|
# Module-specific settings (auto-prefixed with module code)
|
|
api_timeout: int = Field(default=30, description="API timeout in seconds")
|
|
max_retries: int = Field(default=3, description="Max retry attempts")
|
|
|
|
model_config = {"env_prefix": "MYMODULE_"}
|
|
|
|
# Export the config class and instance
|
|
config_class = MyModuleConfig
|
|
config = MyModuleConfig()
|
|
|
|
Usage:
|
|
# Get all module configurations
|
|
from app.modules.config import get_all_module_configs, get_module_config
|
|
|
|
configs = get_all_module_configs()
|
|
# Returns: {"billing": <BillingConfig>, "marketplace": <MarketplaceConfig>, ...}
|
|
|
|
billing_config = get_module_config("billing")
|
|
# Returns: <BillingConfig instance>
|
|
|
|
Integration with ModuleDefinition:
|
|
The config auto-discovery works alongside ModuleDefinition.config_schema.
|
|
- If config.py exists, it's auto-discovered and loaded
|
|
- If config_schema is defined in definition.py, it takes precedence
|
|
- Both can be used together for runtime vs static configuration
|
|
"""
|
|
|
|
import importlib
|
|
import logging
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from pydantic_settings import BaseSettings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Path to the modules directory
|
|
MODULES_DIR = Path(__file__).parent
|
|
|
|
|
|
def discover_module_config(module_code: str) -> "BaseSettings | None":
|
|
"""
|
|
Discover and load configuration for a specific module.
|
|
|
|
Looks for a config.py file in the module directory with either:
|
|
- A 'config' variable (instantiated config)
|
|
- A 'config_class' variable (config class to instantiate)
|
|
|
|
Args:
|
|
module_code: Module code (e.g., "billing", "marketplace")
|
|
|
|
Returns:
|
|
Instantiated config object, or None if no config found
|
|
"""
|
|
# Convert module code to directory name (e.g., dev-tools -> dev_tools)
|
|
dir_name = module_code.replace("-", "_")
|
|
config_file = MODULES_DIR / dir_name / "config.py"
|
|
|
|
if not config_file.exists():
|
|
return None
|
|
|
|
try:
|
|
import_path = f"app.modules.{dir_name}.config"
|
|
config_module = importlib.import_module(import_path)
|
|
|
|
# First, try to get an instantiated config
|
|
if hasattr(config_module, "config"):
|
|
config = config_module.config
|
|
logger.debug(f"Loaded config instance for module {module_code}")
|
|
return config
|
|
|
|
# Otherwise, try to instantiate from config_class
|
|
if hasattr(config_module, "config_class"):
|
|
config_class = config_module.config_class
|
|
config = config_class()
|
|
logger.debug(f"Instantiated config class for module {module_code}")
|
|
return config
|
|
|
|
logger.warning(
|
|
f"Module {module_code} has config.py but no 'config' or 'config_class' export"
|
|
)
|
|
return None
|
|
|
|
except ImportError as e:
|
|
logger.error(f"Failed to import config for module {module_code}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Failed to load config for module {module_code}: {e}")
|
|
return None
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def discover_all_module_configs() -> dict[str, "BaseSettings"]:
|
|
"""
|
|
Discover and load configurations for all modules.
|
|
|
|
Returns:
|
|
Dict mapping module code to config instance
|
|
"""
|
|
configs: dict[str, BaseSettings] = {}
|
|
|
|
for module_dir in sorted(MODULES_DIR.iterdir()):
|
|
if not module_dir.is_dir():
|
|
continue
|
|
|
|
# Skip special directories
|
|
if module_dir.name.startswith("_"):
|
|
continue
|
|
|
|
config_file = module_dir / "config.py"
|
|
if not config_file.exists():
|
|
continue
|
|
|
|
# Convert directory name to module code (e.g., dev_tools -> dev-tools)
|
|
module_code = module_dir.name.replace("_", "-")
|
|
config = discover_module_config(module_code)
|
|
|
|
if config is not None:
|
|
configs[module_code] = config
|
|
|
|
logger.info(f"Discovered configs for {len(configs)} modules")
|
|
return configs
|
|
|
|
|
|
def get_all_module_configs() -> dict[str, "BaseSettings"]:
|
|
"""
|
|
Get all module configurations.
|
|
|
|
Returns:
|
|
Dict mapping module code to config instance
|
|
"""
|
|
return discover_all_module_configs()
|
|
|
|
|
|
def get_module_config(module_code: str) -> "BaseSettings | None":
|
|
"""
|
|
Get configuration for a specific module.
|
|
|
|
Args:
|
|
module_code: Module code (e.g., "billing")
|
|
|
|
Returns:
|
|
Config instance, or None if no config exists
|
|
"""
|
|
configs = get_all_module_configs()
|
|
return configs.get(module_code)
|
|
|
|
|
|
def reload_module_configs() -> dict[str, "BaseSettings"]:
|
|
"""
|
|
Reload all module configurations (clears cache).
|
|
|
|
Useful for testing or when configuration files change.
|
|
|
|
Returns:
|
|
Fresh dict of module configurations
|
|
"""
|
|
discover_all_module_configs.cache_clear()
|
|
return discover_all_module_configs()
|
|
|
|
|
|
def get_merged_config(module_code: str) -> dict[str, Any]:
|
|
"""
|
|
Get merged configuration for a module.
|
|
|
|
Merges configuration from multiple sources:
|
|
1. config.py environment settings (lowest priority)
|
|
2. ModuleDefinition.default_config
|
|
3. ModuleDefinition.config_schema validation
|
|
|
|
Args:
|
|
module_code: Module code
|
|
|
|
Returns:
|
|
Merged configuration dict
|
|
"""
|
|
# Import here to avoid circular imports
|
|
from app.modules.registry import get_module
|
|
|
|
config_dict: dict[str, Any] = {}
|
|
|
|
# 1. Load from config.py if exists
|
|
config_instance = get_module_config(module_code)
|
|
if config_instance is not None:
|
|
config_dict.update(config_instance.model_dump())
|
|
|
|
# 2. Merge with ModuleDefinition defaults
|
|
module = get_module(module_code)
|
|
if module:
|
|
# default_config is merged, overriding config.py values
|
|
if module.default_config:
|
|
for key, value in module.default_config.items():
|
|
if key not in config_dict:
|
|
config_dict[key] = value
|
|
|
|
# 3. Validate against config_schema if defined
|
|
if module.config_schema:
|
|
try:
|
|
validated = module.config_schema(**config_dict)
|
|
config_dict = validated.model_dump()
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"Config validation failed for module {module_code}: {e}"
|
|
)
|
|
|
|
return config_dict
|
|
|
|
|
|
def validate_module_configs() -> list[str]:
|
|
"""
|
|
Validate all module configurations.
|
|
|
|
Returns:
|
|
List of validation errors (empty if all valid)
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
for module_dir in MODULES_DIR.iterdir():
|
|
if not module_dir.is_dir() or module_dir.name.startswith("_"):
|
|
continue
|
|
|
|
config_file = module_dir / "config.py"
|
|
if not config_file.exists():
|
|
continue
|
|
|
|
module_code = module_dir.name.replace("_", "-")
|
|
|
|
try:
|
|
config = discover_module_config(module_code)
|
|
if config is None:
|
|
errors.append(
|
|
f"Module {module_code}: config.py exists but no config exported"
|
|
)
|
|
except Exception as e:
|
|
errors.append(f"Module {module_code}: config.py failed to load: {e}")
|
|
|
|
return errors
|
|
|
|
|
|
__all__ = [
|
|
"discover_module_config",
|
|
"discover_all_module_configs",
|
|
"get_all_module_configs",
|
|
"get_module_config",
|
|
"reload_module_configs",
|
|
"get_merged_config",
|
|
"validate_module_configs",
|
|
]
|