233 lines
9.0 KiB
Python
233 lines
9.0 KiB
Python
import os
|
||
import logging
|
||
import yaml
|
||
|
||
log = logging.getLogger("ainventory")
|
||
|
||
_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),
|
||
"ALLOWED_ORIGINS": (("application", "cors_origins"), str),
|
||
"SERVER_IP": (("application", "server_ip"), str),
|
||
"FRONTEND_PORT": (("application", "frontend_port"), str),
|
||
"FRONTEND_SSL_PORT": (("application", "frontend_ssl_port"), str),
|
||
"BACKEND_PORT": (("application", "backend_port"), str),
|
||
"BACKEND_SSL_PORT": (("application", "backend_ssl_port"), 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")
|
||
|
||
# 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",
|
||
"server_ip": "localhost",
|
||
"frontend_port": "8917",
|
||
"frontend_ssl_port": "8919",
|
||
"backend_port": "8918",
|
||
"backend_ssl_port": "8918"
|
||
},
|
||
"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.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
|
||
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}")
|