#!/usr/bin/env python3
"""
Configuration Manager
Provides unified configuration management with support for environment variables,
YAML files, and database storage.
"""
import os
import yaml
from pathlib import Path
from typing import Any, Dict, List, Optional, Union, Type, TypeVar
from dataclasses import dataclass, field, asdict
from contextlib import contextmanager
from ..utils import get_logger, get_env_bool, get_env_int, get_env_float, get_env_list
from ..database.models import Configuration
from ..database.manager import DatabaseManager
T = TypeVar('T')
logger = get_logger(__name__)
@dataclass
class ServerConfig:
"""Server configuration."""
host: str = '127.0.0.1'
port: int = 8000
debug: bool = False
reload: bool = False
workers: int = 1
max_connections: int = 100
timeout: int = 30
@classmethod
def from_env(cls) -> 'ServerConfig':
"""Create from environment variables."""
return cls(
host=os.getenv('SERVER_HOST', '127.0.0.1'),
port=get_env_int('SERVER_PORT', 8000),
debug=get_env_bool('DEBUG', False),
reload=get_env_bool('RELOAD', False),
workers=get_env_int('WORKERS', 1),
max_connections=get_env_int('MAX_CONNECTIONS', 100),
timeout=get_env_int('TIMEOUT', 30),
)
@dataclass
class DatabaseConfig:
"""Database configuration."""
type: str = 'sqlite'
url: str = 'sqlite:///anydocs.db'
pool_size: int = 5
max_overflow: int = 10
pool_timeout: int = 30
pool_recycle: int = 3600
echo: bool = False
@classmethod
def from_env(cls) -> 'DatabaseConfig':
"""Create from environment variables."""
return cls(
type=os.getenv('DATABASE_TYPE', 'sqlite'),
url=os.getenv('DATABASE_URL', 'sqlite:///anydocs.db'),
pool_size=get_env_int('DATABASE_POOL_SIZE', 5),
max_overflow=get_env_int('DATABASE_MAX_OVERFLOW', 10),
pool_timeout=get_env_int('DATABASE_POOL_TIMEOUT', 30),
pool_recycle=get_env_int('DATABASE_POOL_RECYCLE', 3600),
echo=get_env_bool('DATABASE_ECHO', False),
)
@dataclass
class AuthConfig:
"""Authentication configuration."""
methods: List[str] = field(default_factory=lambda: ['api_key'])
jwt_secret: str = ''
jwt_algorithm: str = 'HS256'
jwt_expiration: int = 3600
api_key_prefix: str = 'ak'
oauth2_providers: Dict[str, Dict[str, str]] = field(default_factory=dict)
@classmethod
def from_env(cls) -> 'AuthConfig':
"""Create from environment variables."""
return cls(
methods=get_env_list('AUTH_METHODS', default=['api_key']),
jwt_secret=os.getenv('JWT_SECRET', ''),
jwt_algorithm=os.getenv('JWT_ALGORITHM', 'HS256'),
jwt_expiration=get_env_int('JWT_EXPIRATION', 3600),
api_key_prefix=os.getenv('API_KEY_PREFIX', 'ak'),
oauth2_providers={}, # Load from config file
)
@dataclass
class CacheConfig:
"""Cache configuration."""
enabled: bool = True
type: str = 'memory'
ttl: int = 3600
max_size: int = 1000
redis_url: str = ''
@classmethod
def from_env(cls) -> 'CacheConfig':
"""Create from environment variables."""
return cls(
enabled=get_env_bool('CACHE_ENABLED', True),
type=os.getenv('CACHE_TYPE', 'memory'),
ttl=get_env_int('CACHE_TTL', 3600),
max_size=get_env_int('CACHE_MAX_SIZE', 1000),
redis_url=os.getenv('REDIS_URL', ''),
)
@dataclass
class LoggingConfig:
"""Logging configuration."""
level: str = 'INFO'
format: str = '%(asctime)s [%(levelname)8s] %(name)s: %(message)s'
file: str = ''
max_size: int = 10 * 1024 * 1024 # 10MB
backup_count: int = 5
json_logs: bool = False
@classmethod
def from_env(cls) -> 'LoggingConfig':
"""Create from environment variables."""
return cls(
level=os.getenv('LOG_LEVEL', 'INFO'),
format=os.getenv('LOG_FORMAT', '%(asctime)s [%(levelname)8s] %(name)s: %(message)s'),
file=os.getenv('LOG_FILE', ''),
max_size=get_env_int('LOG_MAX_SIZE', 10 * 1024 * 1024),
backup_count=get_env_int('LOG_BACKUP_COUNT', 5),
json_logs=get_env_bool('JSON_LOGS', False),
)
@dataclass
class ContentConfig:
"""Content processing configuration."""
max_file_size: int = 50 * 1024 * 1024 # 50MB
supported_formats: List[str] = field(default_factory=lambda: ['markdown', 'html', 'text'])
image_max_size: int = 5 * 1024 * 1024 # 5MB
image_formats: List[str] = field(default_factory=lambda: ['jpg', 'jpeg', 'png', 'gif', 'webp'])
enable_syntax_highlighting: bool = True
@classmethod
def from_env(cls) -> 'ContentConfig':
"""Create from environment variables."""
return cls(
max_file_size=get_env_int('CONTENT_MAX_FILE_SIZE', 50 * 1024 * 1024),
supported_formats=get_env_list('CONTENT_SUPPORTED_FORMATS', default=['markdown', 'html', 'text']),
image_max_size=get_env_int('CONTENT_IMAGE_MAX_SIZE', 5 * 1024 * 1024),
image_formats=get_env_list('CONTENT_IMAGE_FORMATS', default=['jpg', 'jpeg', 'png', 'gif', 'webp']),
enable_syntax_highlighting=get_env_bool('CONTENT_ENABLE_SYNTAX_HIGHLIGHTING', True),
)
@dataclass
class MCPConfig:
"""MCP server configuration."""
name: str = 'AnyDocs MCP Server'
version: str = '1.0.0'
description: str = 'Transform documentation into MCP-compatible server'
max_resources: int = 1000
max_tools: int = 50
enable_discovery: bool = True
@classmethod
def from_env(cls) -> 'MCPConfig':
"""Create from environment variables."""
return cls(
name=os.getenv('MCP_NAME', 'AnyDocs MCP Server'),
version=os.getenv('MCP_VERSION', '1.0.0'),
description=os.getenv('MCP_DESCRIPTION', 'Transform documentation into MCP-compatible server'),
max_resources=get_env_int('MCP_MAX_RESOURCES', 1000),
max_tools=get_env_int('MCP_MAX_TOOLS', 50),
enable_discovery=get_env_bool('MCP_ENABLE_DISCOVERY', True),
)
@dataclass
class AppConfig:
"""Main application configuration."""
server: ServerConfig = field(default_factory=ServerConfig)
database: DatabaseConfig = field(default_factory=DatabaseConfig)
auth: AuthConfig = field(default_factory=AuthConfig)
cache: CacheConfig = field(default_factory=CacheConfig)
logging: LoggingConfig = field(default_factory=LoggingConfig)
content: ContentConfig = field(default_factory=ContentConfig)
mcp: MCPConfig = field(default_factory=MCPConfig)
@classmethod
def from_env(cls) -> 'AppConfig':
"""Create from environment variables."""
return cls(
server=ServerConfig.from_env(),
database=DatabaseConfig.from_env(),
auth=AuthConfig.from_env(),
cache=CacheConfig.from_env(),
logging=LoggingConfig.from_env(),
content=ContentConfig.from_env(),
mcp=MCPConfig.from_env(),
)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'AppConfig':
"""Create from dictionary."""
return cls(
server=ServerConfig(**data.get('server', {})),
database=DatabaseConfig(**data.get('database', {})),
auth=AuthConfig(**data.get('auth', {})),
cache=CacheConfig(**data.get('cache', {})),
logging=LoggingConfig(**data.get('logging', {})),
content=ContentConfig(**data.get('content', {})),
mcp=MCPConfig(**data.get('mcp', {})),
)
class ConfigManager:
"""Configuration manager with multiple sources."""
def __init__(
self,
config_file: Optional[str] = None,
db_manager: Optional[DatabaseManager] = None,
auto_reload: bool = False
):
"""
Initialize configuration manager.
Args:
config_file: Path to YAML configuration file
db_manager: Database manager for persistent config
auto_reload: Enable automatic config reloading
"""
self.config_file = config_file
self.db_manager = db_manager
self.auto_reload = auto_reload
self._config: Optional[AppConfig] = None
self._file_mtime: Optional[float] = None
logger.info("Configuration manager initialized", config_file=config_file)
def load_config(self, force_reload: bool = False) -> AppConfig:
"""Load configuration from all sources.
Args:
force_reload: Force reload even if cached
Returns:
Application configuration
"""
if self._config is not None and not force_reload and not self._should_reload():
return self._config
logger.info("Loading configuration")
# Start with environment variables
config = AppConfig.from_env()
# Override with file configuration
if self.config_file:
file_config = self._load_file_config()
if file_config:
config = self._merge_configs(config, file_config)
# Override with database configuration
if self.db_manager:
db_config = self._load_db_config()
if db_config:
config = self._merge_configs(config, db_config)
self._config = config
self._update_file_mtime()
logger.info("Configuration loaded successfully")
return config
def save_config(self, config: AppConfig, target: str = 'file') -> bool:
"""Save configuration to target.
Args:
config: Configuration to save
target: Target ('file' or 'database')
Returns:
True if successful
"""
try:
if target == 'file' and self.config_file:
return self._save_file_config(config)
elif target == 'database' and self.db_manager:
return self._save_db_config(config)
else:
logger.error("Invalid save target or missing manager", target=target)
return False
except Exception as e:
logger.error("Failed to save configuration", target=target, error=str(e))
return False
def get_config(self) -> AppConfig:
"""Get current configuration.
Returns:
Current application configuration
"""
return self.load_config()
def update_config(self, updates: Dict[str, Any], save_to: Optional[str] = None) -> bool:
"""Update configuration with partial updates.
Args:
updates: Configuration updates
save_to: Target to save to ('file', 'database', or None)
Returns:
True if successful
"""
try:
current_config = self.get_config()
config_dict = current_config.to_dict()
# Apply updates
self._deep_update(config_dict, updates)
# Create new config
new_config = AppConfig.from_dict(config_dict)
# Save if requested
if save_to:
if not self.save_config(new_config, save_to):
return False
# Update cached config
self._config = new_config
logger.info("Configuration updated", updates=updates)
return True
except Exception as e:
logger.error("Failed to update configuration", error=str(e))
return False
def get_setting(self, key: str, default: Any = None) -> Any:
"""Get specific setting value.
Args:
key: Setting key (dot notation supported)
default: Default value if not found
Returns:
Setting value
"""
config = self.get_config()
config_dict = config.to_dict()
# Navigate through nested keys
keys = key.split('.')
value = config_dict
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set_setting(self, key: str, value: Any, save_to: Optional[str] = None) -> bool:
"""Set specific setting value.
Args:
key: Setting key (dot notation supported)
value: Setting value
save_to: Target to save to
Returns:
True if successful
"""
keys = key.split('.')
updates = {}
# Build nested update dictionary
current = updates
for k in keys[:-1]:
current[k] = {}
current = current[k]
current[keys[-1]] = value
return self.update_config(updates, save_to)
def _load_file_config(self) -> Optional[AppConfig]:
"""Load configuration from YAML file."""
if not self.config_file or not Path(self.config_file).exists():
return None
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
data = yaml.safe_load(f)
if not data:
return None
return AppConfig.from_dict(data)
except Exception as e:
logger.error("Failed to load file configuration", file=self.config_file, error=str(e))
return None
def _save_file_config(self, config: AppConfig) -> bool:
"""Save configuration to YAML file."""
if not self.config_file:
return False
try:
# Ensure directory exists
Path(self.config_file).parent.mkdir(parents=True, exist_ok=True)
with open(self.config_file, 'w', encoding='utf-8') as f:
yaml.dump(config.to_dict(), f, default_flow_style=False, indent=2)
self._update_file_mtime()
logger.info("Configuration saved to file", file=self.config_file)
return True
except Exception as e:
logger.error("Failed to save file configuration", file=self.config_file, error=str(e))
return False
def _load_db_config(self) -> Optional[AppConfig]:
"""Load configuration from database."""
if not self.db_manager:
return None
try:
with self.db_manager.get_session() as session:
configs = session.query(Configuration).all()
if not configs:
return None
# Build config dictionary
config_dict = {}
for config in configs:
keys = config.key.split('.')
current = config_dict
for key in keys[:-1]:
if key not in current:
current[key] = {}
current = current[key]
# Parse value based on type
if config.value_type == 'int':
current[keys[-1]] = int(config.value)
elif config.value_type == 'float':
current[keys[-1]] = float(config.value)
elif config.value_type == 'bool':
current[keys[-1]] = config.value.lower() == 'true'
elif config.value_type == 'list':
current[keys[-1]] = config.value.split(',')
else:
current[keys[-1]] = config.value
return AppConfig.from_dict(config_dict)
except Exception as e:
logger.error("Failed to load database configuration", error=str(e))
return None
def _save_db_config(self, config: AppConfig) -> bool:
"""Save configuration to database."""
if not self.db_manager:
return False
try:
with self.db_manager.get_session() as session:
# Clear existing configuration
session.query(Configuration).delete()
# Flatten config dictionary
flat_config = self._flatten_dict(config.to_dict())
# Save each setting
for key, value in flat_config.items():
value_type = 'str'
if isinstance(value, int):
value_type = 'int'
elif isinstance(value, float):
value_type = 'float'
elif isinstance(value, bool):
value_type = 'bool'
elif isinstance(value, list):
value_type = 'list'
value = ','.join(str(v) for v in value)
config_obj = Configuration(
key=key,
value=str(value),
value_type=value_type,
description=f"Configuration setting: {key}"
)
session.add(config_obj)
session.commit()
logger.info("Configuration saved to database")
return True
except Exception as e:
logger.error("Failed to save database configuration", error=str(e))
return False
def _merge_configs(self, base: AppConfig, override: AppConfig) -> AppConfig:
"""Merge two configurations."""
base_dict = base.to_dict()
override_dict = override.to_dict()
self._deep_update(base_dict, override_dict)
return AppConfig.from_dict(base_dict)
def _deep_update(self, base: Dict[str, Any], updates: Dict[str, Any]) -> None:
"""Deep update dictionary."""
for key, value in updates.items():
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
self._deep_update(base[key], value)
else:
base[key] = value
def _flatten_dict(self, d: Dict[str, Any], parent_key: str = '', sep: str = '.') -> Dict[str, Any]:
"""Flatten nested dictionary."""
items = []
for k, v in d.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
items.extend(self._flatten_dict(v, new_key, sep=sep).items())
else:
items.append((new_key, v))
return dict(items)
def _should_reload(self) -> bool:
"""Check if configuration should be reloaded."""
if not self.auto_reload or not self.config_file:
return False
try:
current_mtime = Path(self.config_file).stat().st_mtime
return self._file_mtime is None or current_mtime > self._file_mtime
except Exception:
return False
def _update_file_mtime(self) -> None:
"""Update file modification time."""
if self.config_file:
try:
self._file_mtime = Path(self.config_file).stat().st_mtime
except Exception:
self._file_mtime = None
@contextmanager
def config_context(self, **overrides):
"""Context manager for temporary configuration overrides."""
original_config = self._config
try:
if overrides:
current_config = self.get_config()
config_dict = current_config.to_dict()
self._deep_update(config_dict, overrides)
self._config = AppConfig.from_dict(config_dict)
yield self._config
finally:
self._config = original_config
# Global configuration manager instance
_config_manager: Optional[ConfigManager] = None
def init_config_manager(
config_file: Optional[str] = None,
db_manager: Optional[DatabaseManager] = None,
auto_reload: bool = False
) -> ConfigManager:
"""Initialize global configuration manager.
Args:
config_file: Path to configuration file
db_manager: Database manager instance
auto_reload: Enable automatic reloading
Returns:
ConfigManager instance
"""
global _config_manager
_config_manager = ConfigManager(config_file, db_manager, auto_reload)
return _config_manager
def get_config_manager() -> Optional[ConfigManager]:
"""Get global configuration manager.
Returns:
ConfigManager instance or None
"""
return _config_manager
def get_config() -> AppConfig:
"""Get current application configuration.
Returns:
Application configuration
"""
if _config_manager:
return _config_manager.get_config()
else:
# Fallback to environment-only config
return AppConfig.from_env()
def get_setting(key: str, default: Any = None) -> Any:
"""Get configuration setting.
Args:
key: Setting key
default: Default value
Returns:
Setting value
"""
if _config_manager:
return _config_manager.get_setting(key, default)
else:
# Fallback to environment variables
env_key = key.upper().replace('.', '_')
return os.getenv(env_key, default)
