268 lines
10 KiB
Python
268 lines
10 KiB
Python
|
"""
|
||
|
|
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
|