# 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// └── 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_", "env_file": ".env", "extra": "ignore"} # 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": , "marketplace": , ...} billing_config = get_module_config("billing") # Returns: 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", ]