config_loader.py•8.14 kB
"""
Production-grade configuration loader
Loads from YAML, environment variables, and provides validation
"""
import os
import yaml
from pathlib import Path
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings
import structlog
from string import Template
logger = structlog.get_logger(__name__)
PROJECT_ROOT = Path(__file__).parent.parent.parent
class Config(BaseSettings):
"""
Application configuration with environment variable support
Environment variables override config file values
"""
# Environment
environment: str = Field("production", env="ENVIRONMENT")
debug: bool = Field(False, env="DEBUG")
# Google API
google_client_id: str = Field(..., env="GOOGLE_CLIENT_ID")
google_client_secret: str = Field(..., env="GOOGLE_CLIENT_SECRET")
google_redirect_uri: str = Field("http://localhost:8080", env="GOOGLE_REDIRECT_URI")
# LLM Providers
euron_api_key: Optional[str] = Field(None, env="EURON_API_KEY")
euron_api_base: str = Field("https://api.euron.one/api/v1/euri", env="EURON_API_BASE")
euron_model: str = Field("gpt-4.1-nano", env="EURON_MODEL")
deepseek_api_key: Optional[str] = Field(None, env="DEEPSEEK_API_KEY")
deepseek_api_base: str = Field("https://api.deepseek.com/v1", env="DEEPSEEK_API_BASE")
deepseek_model: str = Field("deepseek-chat", env="DEEPSEEK_MODEL")
gemini_api_key: Optional[str] = Field(None, env="GEMINI_API_KEY")
gemini_model: str = Field("gemini-pro", env="GEMINI_MODEL")
anthropic_api_key: Optional[str] = Field(None, env="ANTHROPIC_API_KEY")
anthropic_model: str = Field("claude-3-5-sonnet-20241022", env="ANTHROPIC_MODEL")
# Application settings
email_summary_recipient: str = Field(..., env="EMAIL_SUMMARY_RECIPIENT")
email_summary_timezone: str = Field("America/New_York", env="EMAIL_SUMMARY_TIMEZONE")
calendar_timezone: str = Field("America/New_York", env="CALENDAR_TIMEZONE")
# Database
database_url: str = Field(
"sqlite+aiosqlite:///./data/enhanced_mcp.db",
env="DATABASE_URL"
)
# Logging
log_level: str = Field("INFO", env="LOG_LEVEL")
log_format: str = Field("json", env="LOG_FORMAT")
log_file: str = Field("logs/enhanced_mcp.log", env="LOG_FILE")
# Monitoring
monitoring_enabled: bool = Field(True, env="MONITORING_ENABLED")
metrics_port: int = Field(9090, env="METRICS_PORT")
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
extra = "ignore" # Allow extra fields from .env file
extra = "ignore" # Allow extra fields from .env file
class ConfigLoader:
"""
Configuration loader with YAML and environment variable support
Features:
- Load from YAML file
- Environment variable substitution
- Validation
- Singleton pattern
"""
def __init__(self, config_path: Optional[Path] = None):
self.config_path = config_path or (PROJECT_ROOT / "config" / "config.yaml")
self._yaml_config: Dict[str, Any] = {}
self._env_config: Optional[Config] = None
self.load()
def load(self):
"""Load configuration from all sources"""
# Load YAML config
if self.config_path.exists():
try:
with open(self.config_path, 'r') as f:
raw_config = f.read()
# Substitute environment variables in YAML
raw_config = self._substitute_env_vars(raw_config)
self._yaml_config = yaml.safe_load(raw_config) or {}
logger.info(
"Loaded YAML configuration",
path=str(self.config_path)
)
except Exception as e:
logger.error(
"Failed to load YAML configuration",
error=str(e),
path=str(self.config_path)
)
self._yaml_config = {}
else:
logger.warning(
"Config file not found, using environment variables only",
path=str(self.config_path)
)
# Load environment config
try:
self._env_config = Config()
logger.info("Loaded environment configuration")
except Exception as e:
logger.error(
"Failed to load environment configuration",
error=str(e)
)
raise
def _substitute_env_vars(self, content: str) -> str:
"""Substitute environment variables in format ${VAR_NAME}"""
template = Template(content)
# Create substitution dict with all environment variables
env_dict = os.environ.copy()
try:
return template.safe_substitute(env_dict)
except Exception as e:
logger.warning(
"Failed to substitute some environment variables",
error=str(e)
)
return content
def get(self, key: str, default: Any = None) -> Any:
"""
Get configuration value
Precedence: Environment variable > YAML config > Default
Args:
key: Configuration key (dot notation for nested keys)
default: Default value if not found
Returns:
Configuration value
"""
# Try environment config first
if self._env_config and hasattr(self._env_config, key.lower()):
return getattr(self._env_config, key.lower())
# Try YAML config (support dot notation)
value = self._yaml_config
for part in key.split('.'):
if isinstance(value, dict) and part in value:
value = value[part]
else:
return default
return value if value != self._yaml_config else default
def get_section(self, section: str) -> Dict[str, Any]:
"""Get entire configuration section"""
return self._yaml_config.get(section, {})
def get_llm_config(self) -> Dict[str, Any]:
"""Get LLM provider configuration with credentials"""
llm_config = self.get_section("llm")
# Inject API keys from environment
if "providers" in llm_config:
for provider in llm_config["providers"]:
name = provider["name"]
if name == "euron":
provider["api_key"] = self._env_config.euron_api_key
provider["api_base"] = self._env_config.euron_api_base
provider["model"] = self._env_config.euron_model
elif name == "deepseek":
provider["api_key"] = self._env_config.deepseek_api_key
provider["api_base"] = self._env_config.deepseek_api_base
provider["model"] = self._env_config.deepseek_model
elif name == "gemini":
provider["api_key"] = self._env_config.gemini_api_key
provider["model"] = self._env_config.gemini_model
elif name == "claude":
provider["api_key"] = self._env_config.anthropic_api_key
provider["model"] = self._env_config.anthropic_model
return llm_config
def reload(self):
"""Reload configuration"""
logger.info("Reloading configuration")
self.load()
@property
def yaml_config(self) -> Dict[str, Any]:
"""Get raw YAML configuration"""
return self._yaml_config
@property
def env_config(self) -> Config:
"""Get environment configuration"""
return self._env_config
# Singleton instance
_config_loader: Optional[ConfigLoader] = None
def get_config() -> ConfigLoader:
"""Get singleton configuration loader"""
global _config_loader
if _config_loader is None:
_config_loader = ConfigLoader()
return _config_loader
def reload_config():
"""Reload configuration"""
global _config_loader
if _config_loader:
_config_loader.reload()