""" Enhanced configuration loader with comprehensive environment variable expansion Implements ${VAR} and ${VAR:-default} pattern support with recursive expansion """ import os import re from typing import Any, Dict, List, Union from pathlib import Path try: from .dependency_manager import get_dependency_manager except ImportError: from dependency_manager import get_dependency_manager # Try to import dependencies with graceful degradation dependency_manager = get_dependency_manager() yaml = dependency_manager.get_module('yaml') class ConfigurationError(Exception): """Base exception for configuration-related errors""" pass class EnvironmentVariableError(ConfigurationError): """Exception for environment variable expansion errors""" pass class ConfigurationLoader: """Enhanced configuration loader with environment variable expansion""" def __init__(self): # Pattern to match ${VAR} and ${VAR:-default} syntax self.env_var_pattern = re.compile(r'\$\{([^}]+)\}') def expand_environment_variables(self, config_dict: Dict[str, Any]) -> Dict[str, Any]: """ Recursively expand environment variables in configuration Supports patterns: - ${VAR}: Required variable, raises error if not set - ${VAR:-default}: Variable with default value - ${VAR:?error_message}: Required variable with custom error message Args: config_dict: Configuration dictionary to expand Returns: Configuration dictionary with expanded environment variables Raises: EnvironmentVariableError: If required environment variable is missing """ return self._expand_value(config_dict) def _expand_value(self, value: Any) -> Any: """Recursively expand environment variables in any value type""" if isinstance(value, str): return self._expand_string_env_vars(value) elif isinstance(value, dict): return {k: self._expand_value(v) for k, v in value.items()} elif isinstance(value, list): return [self._expand_value(item) for item in value] else: return value def _expand_string_env_vars(self, text: str) -> str: """ Expand environment variables in a string with enhanced support Supports multiple expansion patterns: - ${VAR}: Required variable - ${VAR:-default}: Variable with default value - ${VAR:?error_message}: Required variable with custom error message Args: text: String potentially containing environment variable references Returns: String with environment variables expanded Raises: EnvironmentVariableError: If required environment variable is missing """ def replace_env_var(match): var_expr = match.group(1) # Handle ${VAR:-default} pattern if ':-' in var_expr: var_name, default_value = var_expr.split(':-', 1) var_name = var_name.strip() env_value = os.getenv(var_name) return env_value if env_value is not None else default_value # Handle ${VAR:?error_message} pattern elif ':?' in var_expr: var_name, error_msg = var_expr.split(':?', 1) var_name = var_name.strip() env_value = os.getenv(var_name) if env_value is None: raise EnvironmentVariableError( f"Environment variable '{var_name}' is required: {error_msg}" ) return env_value # Handle simple ${VAR} pattern else: var_name = var_expr.strip() env_value = os.getenv(var_name) if env_value is None: raise EnvironmentVariableError( f"Environment variable '{var_name}' is not set. " f"Please set it or provide a default value using ${{{var_name}:-default}}" ) return env_value return self.env_var_pattern.sub(replace_env_var, text) @classmethod def load_config(cls, config_path: Union[str, Path]) -> Dict[str, Any]: """ Load and validate configuration with environment variable expansion Args: config_path: Path to configuration file (YAML or JSON) Returns: Configuration dictionary with expanded environment variables Raises: ConfigurationError: If configuration file cannot be loaded or parsed EnvironmentVariableError: If required environment variables are missing """ loader = cls() config_path = Path(config_path) if not config_path.exists(): raise ConfigurationError(f"Configuration file not found: {config_path}") # Load raw configuration try: with config_path.open('r', encoding='utf-8') as f: if config_path.suffix.lower() in ['.yaml', '.yml']: if yaml is None: raise ConfigurationError( 'YAML configuration files require PyYAML library.\n' 'Please install it with: pip install pyyaml>=6.0' ) config_dict = yaml.safe_load(f) or {} elif config_path.suffix.lower() == '.json': import json config_dict = json.load(f) else: raise ConfigurationError( 'Configuration file must be YAML (.yaml/.yml) or JSON (.json)' ) except Exception as e: if isinstance(e, ConfigurationError): raise elif yaml is None and 'yaml' in str(e).lower(): raise ConfigurationError( 'YAML parsing failed - PyYAML library not available.\n' 'Please install it with: pip install pyyaml>=6.0\n' f'Original error: {e}' ) else: raise ConfigurationError(f"Failed to parse configuration file: {e}") # Expand environment variables try: expanded_config = loader.expand_environment_variables(config_dict) return expanded_config except EnvironmentVariableError as e: raise EnvironmentVariableError(f"Environment variable expansion failed: {e}") def validate_environment_variables(self, config_dict: Dict[str, Any]) -> List[str]: """ Validate that all required environment variables are available Args: config_dict: Configuration dictionary to validate Returns: List of missing environment variable error messages """ missing_vars = [] def check_env_vars_in_value(value: Any, path: str = "") -> None: if isinstance(value, str): # Find all environment variable references matches = self.env_var_pattern.findall(value) for match in matches: var_expr = match # Skip variables with default values if ':-' in var_expr: continue # Handle custom error message pattern if ':?' in var_expr: var_name = var_expr.split(':?')[0].strip() else: var_name = var_expr.strip() if not os.getenv(var_name): location = f" at {path}" if path else "" missing_vars.append(f"Environment variable '{var_name}'{location} is not set") elif isinstance(value, dict): for key, val in value.items(): new_path = f"{path}.{key}" if path else key check_env_vars_in_value(val, new_path) elif isinstance(value, list): for i, item in enumerate(value): new_path = f"{path}[{i}]" if path else f"[{i}]" check_env_vars_in_value(item, new_path) check_env_vars_in_value(config_dict) return missing_vars def get_environment_variable_references(self, config_dict: Dict[str, Any]) -> Dict[str, List[str]]: """ Get all environment variable references in the configuration Args: config_dict: Configuration dictionary to analyze Returns: Dictionary mapping variable names to list of locations where they're used """ env_vars = {} def collect_env_vars_in_value(value: Any, path: str = "") -> None: if isinstance(value, str): matches = self.env_var_pattern.findall(value) for match in matches: var_expr = match # Extract variable name (handle default value and error message patterns) if ':-' in var_expr: var_name = var_expr.split(':-')[0].strip() elif ':?' in var_expr: var_name = var_expr.split(':?')[0].strip() else: var_name = var_expr.strip() if var_name not in env_vars: env_vars[var_name] = [] location = path if path else "root" if location not in env_vars[var_name]: env_vars[var_name].append(location) elif isinstance(value, dict): for key, val in value.items(): new_path = f"{path}.{key}" if path else key collect_env_vars_in_value(val, new_path) elif isinstance(value, list): for i, item in enumerate(value): new_path = f"{path}[{i}]" if path else f"[{i}]" collect_env_vars_in_value(item, new_path) collect_env_vars_in_value(config_dict) return env_vars