feat(07-02): refactor config_loader.py for YAML and D-06 load order
- Replace dotenv with PyYAML - Implement D-06 load order: env vars > YAML > defaults - Add validation and logging for config - Remove inventory.env references - Update requirements.txt
This commit is contained in:
@@ -1,34 +1,221 @@
|
||||
import os
|
||||
import logging
|
||||
from dotenv import load_dotenv
|
||||
import yaml
|
||||
|
||||
log = logging.getLogger("ainventory")
|
||||
|
||||
def load_config():
|
||||
"""
|
||||
Centralized environment loader for TFM aInventory.
|
||||
Prioritizes existing environment variables (Docker),
|
||||
then inventory.env at project root, then backend/.env.
|
||||
"""
|
||||
_config = {}
|
||||
|
||||
class ConfigError(Exception):
|
||||
"""Raised when configuration is invalid or missing."""
|
||||
pass
|
||||
|
||||
def _deep_merge(base, source):
|
||||
"""Recursively merge dictionaries."""
|
||||
for key, value in source.items():
|
||||
if isinstance(value, dict) and key in base and isinstance(base[key], dict):
|
||||
_deep_merge(base[key], value)
|
||||
else:
|
||||
base[key] = value
|
||||
|
||||
def _map_secrets(config, secrets):
|
||||
"""Map secrets from secrets.yaml (flattened) to the nested config structure."""
|
||||
mapping = {
|
||||
"JWT_SECRET_KEY": ("auth", "jwt_secret_key"),
|
||||
"GEMINI_API_KEY": ("ai", "gemini_api_key"),
|
||||
"CLAUDE_API_KEY": ("ai", "claude_api_key"),
|
||||
"DATABASE_PASSWORD": ("database", "password"),
|
||||
"LDAP_PASSWORD": ("auth", "ldap_password")
|
||||
}
|
||||
for secret_key, config_path in mapping.items():
|
||||
if secret_key in secrets:
|
||||
target = config
|
||||
for p in config_path[:-1]:
|
||||
if p not in target:
|
||||
target[p] = {}
|
||||
target = target[p]
|
||||
target[config_path[-1]] = secrets[secret_key]
|
||||
|
||||
def _to_bool(val):
|
||||
"""Convert string to boolean."""
|
||||
if isinstance(val, bool):
|
||||
return val
|
||||
return str(val).lower() in ("true", "1", "yes", "on")
|
||||
|
||||
def _apply_env_overrides(config):
|
||||
"""Apply environment variable overrides (D-06 load order)."""
|
||||
# Mapping of environment variables to config paths
|
||||
# Format: ENV_VAR: (path, type_converter)
|
||||
env_mapping = {
|
||||
"BACKEND_DATABASE_SQLITE_PATH": (("database", "sqlite_path"), str),
|
||||
"BACKEND_DATABASE_WAL_MODE": (("database", "wal_mode"), _to_bool),
|
||||
"BACKEND_DATABASE_LOG_RETENTION_DAYS": (("database", "log_retention_days"), int),
|
||||
"BACKEND_AI_PRIMARY_AI_PROVIDER": (("ai", "primary_ai_provider"), str),
|
||||
"PRIMARY_AI_PROVIDER": (("ai", "primary_ai_provider"), str),
|
||||
"BACKEND_AI_FALLBACK_PROVIDER": (("ai", "fallback_provider"), str),
|
||||
"BACKEND_AI_GEMINI_API_KEY": (("ai", "gemini_api_key"), str),
|
||||
"GEMINI_API_KEY": (("ai", "gemini_api_key"), str),
|
||||
"BACKEND_AI_CLAUDE_API_KEY": (("ai", "claude_api_key"), str),
|
||||
"CLAUDE_API_KEY": (("ai", "claude_api_key"), str),
|
||||
"BACKEND_AUTH_JWT_SECRET_KEY": (("auth", "jwt_secret_key"), str),
|
||||
"JWT_SECRET_KEY": (("auth", "jwt_secret_key"), str),
|
||||
"BACKEND_AUTH_LDAP_SERVER": (("auth", "ldap_server"), str),
|
||||
"BACKEND_AUTH_LDAP_BASE_DN": (("auth", "ldap_base_dn"), str),
|
||||
"BACKEND_AUTH_PASSWORD_CACHE_PATH": (("auth", "password_cache_path"), str),
|
||||
"BACKEND_LOGGING_LOG_LEVEL": (("logging", "log_level"), str),
|
||||
"LOG_LEVEL": (("logging", "log_level"), str),
|
||||
"BACKEND_LOGGING_LOG_ROTATION_SIZE_MB": (("logging", "log_rotation_size_mb"), int),
|
||||
"BACKEND_LOGGING_LOG_ROTATION_COUNT": (("logging", "log_rotation_count"), int),
|
||||
"BACKEND_APPLICATION_DATA_DIR": (("application", "data_dir"), str),
|
||||
"DATA_DIR": (("application", "data_dir"), str),
|
||||
"BACKEND_APPLICATION_LOGS_DIR": (("application", "logs_dir"), str),
|
||||
"LOGS_DIR": (("application", "logs_dir"), str),
|
||||
"BACKEND_APPLICATION_CORS_ORIGINS": (("application", "cors_origins"), str),
|
||||
"EXTRA_ALLOWED_ORIGINS": (("application", "cors_origins"), str),
|
||||
}
|
||||
|
||||
sensitive_vars = [
|
||||
"JWT_SECRET_KEY", "GEMINI_API_KEY", "CLAUDE_API_KEY",
|
||||
"BACKEND_AUTH_JWT_SECRET_KEY", "BACKEND_AI_GEMINI_API_KEY", "BACKEND_AI_CLAUDE_API_KEY",
|
||||
"LDAP_PASSWORD", "DATABASE_PASSWORD"
|
||||
]
|
||||
|
||||
for env_var, (path, converter) in env_mapping.items():
|
||||
val = os.getenv(env_var)
|
||||
if val is not None:
|
||||
try:
|
||||
converted_val = converter(val)
|
||||
target = config
|
||||
for p in path[:-1]:
|
||||
if p not in target:
|
||||
target[p] = {}
|
||||
target = target[p]
|
||||
|
||||
target[path[-1]] = converted_val
|
||||
|
||||
# Mask sensitive values in logs
|
||||
log_val = "********" if env_var in sensitive_vars else converted_val
|
||||
log.info(f"ℹ️ Override {'.'.join(path)} from environment ({env_var}): {log_val}")
|
||||
except Exception as e:
|
||||
log.warning(f"⚠️ Failed to convert env var {env_var}='{val}': {e}")
|
||||
|
||||
def load_config() -> dict:
|
||||
"""Load config from YAML files with env var overrides (D-06 load order)."""
|
||||
global _config
|
||||
|
||||
# Base directory is backend/
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
# Project root is one level up
|
||||
project_root = os.path.dirname(base_dir)
|
||||
config_dir = os.path.join(project_root, "config")
|
||||
|
||||
inventory_env_path = os.path.join(project_root, "inventory.env")
|
||||
backend_env_path = os.path.join(base_dir, ".env")
|
||||
|
||||
# Check for inventory.env in root (Master Config)
|
||||
if os.path.exists(inventory_env_path):
|
||||
load_dotenv(inventory_env_path)
|
||||
log.info(f"✅ Loaded master configuration from {inventory_env_path}")
|
||||
|
||||
# Check for local backend/.env (Legacy/Fragmented)
|
||||
elif os.path.exists(backend_env_path):
|
||||
load_dotenv(backend_env_path)
|
||||
log.info(f"ℹ️ Loaded local configuration from {backend_env_path}")
|
||||
# 1. Define Defaults
|
||||
config = {
|
||||
"database": {
|
||||
"sqlite_path": "data/inventory.db",
|
||||
"wal_mode": True,
|
||||
"log_retention_days": 30
|
||||
},
|
||||
"ai": {
|
||||
"primary_ai_provider": "gemini",
|
||||
"fallback_provider": "claude",
|
||||
"gemini_api_key": "",
|
||||
"claude_api_key": ""
|
||||
},
|
||||
"auth": {
|
||||
"jwt_secret_key": "change_me_in_production",
|
||||
"ldap_server": "",
|
||||
"ldap_base_dn": "",
|
||||
"password_cache_path": "data/.passwords"
|
||||
},
|
||||
"logging": {
|
||||
"log_level": "INFO",
|
||||
"log_rotation_size_mb": 10,
|
||||
"log_rotation_count": 5
|
||||
},
|
||||
"application": {
|
||||
"data_dir": "./data",
|
||||
"logs_dir": "./logs",
|
||||
"cors_origins": "http://localhost:8917"
|
||||
},
|
||||
"features": {
|
||||
"ai_extraction_enabled": True,
|
||||
"offline_sync_enabled": True,
|
||||
"audit_logging_enabled": True
|
||||
}
|
||||
}
|
||||
|
||||
# 2. Load backend.yaml
|
||||
backend_yaml_path = os.path.join(config_dir, "backend.yaml")
|
||||
if os.path.exists(backend_yaml_path):
|
||||
try:
|
||||
with open(backend_yaml_path, 'r') as f:
|
||||
yaml_data = yaml.safe_load(f) or {}
|
||||
_deep_merge(config, yaml_data)
|
||||
log.info(f"✅ Loaded configuration from {backend_yaml_path}")
|
||||
except Exception as e:
|
||||
log.warning(f"⚠️ Failed to load {backend_yaml_path}: {e}")
|
||||
else:
|
||||
log.info("ℹ️ [CONFIG] Using system environment variables (Docker/Server environment).")
|
||||
log.warning(f"ℹ️ {backend_yaml_path} not found. Using defaults.")
|
||||
|
||||
# 3. Load secrets.yaml
|
||||
secrets_yaml_path = os.path.join(config_dir, "secrets.yaml")
|
||||
if os.path.exists(secrets_yaml_path):
|
||||
try:
|
||||
with open(secrets_yaml_path, 'r') as f:
|
||||
secrets_data = yaml.safe_load(f) or {}
|
||||
_map_secrets(config, secrets_data)
|
||||
log.info(f"✅ Loaded secrets from {secrets_yaml_path}")
|
||||
except Exception as e:
|
||||
log.warning(f"⚠️ Failed to load {secrets_yaml_path}: {e}")
|
||||
|
||||
# 4. Apply Environment Overrides (D-06)
|
||||
_apply_env_overrides(config)
|
||||
|
||||
_config = config
|
||||
return _config
|
||||
|
||||
def get_config() -> dict:
|
||||
"""Get loaded config."""
|
||||
if not _config:
|
||||
load_config()
|
||||
return _config
|
||||
|
||||
def validate_config(config: dict) -> bool:
|
||||
"""Validate config has all required values."""
|
||||
required = [
|
||||
("auth", "jwt_secret_key"),
|
||||
("ai", "primary_ai_provider"),
|
||||
]
|
||||
|
||||
missing = []
|
||||
for path in required:
|
||||
val = config
|
||||
for p in path:
|
||||
val = val.get(p)
|
||||
if not val or val == "change_me_in_production" or val == "CHANGE_ME_IN_PRODUCTION_MIN_32_CHARS":
|
||||
missing.append(".".join(path))
|
||||
|
||||
if missing:
|
||||
raise ConfigError(f"Missing or default required configuration: {', '.join(missing)}")
|
||||
|
||||
# Validate enums
|
||||
valid_providers = ["gemini", "claude"]
|
||||
if config["ai"]["primary_ai_provider"] not in valid_providers:
|
||||
raise ConfigError(f"Invalid primary_ai_provider: {config['ai']['primary_ai_provider']}. Must be one of {valid_providers}")
|
||||
|
||||
valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR"]
|
||||
if config["logging"]["log_level"].upper() not in valid_log_levels:
|
||||
raise ConfigError(f"Invalid log_level: {config['logging']['log_level']}. Must be one of {valid_log_levels}")
|
||||
|
||||
log.info(f"✅ Config validated: primary_ai_provider={config['ai']['primary_ai_provider']}, log_level={config['logging']['log_level']}")
|
||||
return True
|
||||
|
||||
# Auto-run if imported
|
||||
load_config()
|
||||
try:
|
||||
load_config()
|
||||
validate_config(_config)
|
||||
except ConfigError as ce:
|
||||
log.error(f"❌ Configuration error: {ce}")
|
||||
except Exception as e:
|
||||
log.error(f"❌ Unexpected error during config load: {e}")
|
||||
|
||||
@@ -5,7 +5,7 @@ pydantic>=2.0.0
|
||||
pydantic-settings>=2.0.0
|
||||
google-genai>=0.1.0
|
||||
anthropic>=0.40.0
|
||||
python-dotenv>=1.0.0
|
||||
PyYAML>=6.0.1
|
||||
Pillow>=10.0.0
|
||||
python-multipart>=0.0.9
|
||||
ldap3>=2.9.1
|
||||
|
||||
Reference in New Issue
Block a user