"""Configuration persistence service for managing application settings. This service handles: - Loading and saving configuration to JSON files - Configuration validation - Backup and restore functionality - Configuration version management """ import json import shutil from datetime import datetime from pathlib import Path from typing import Dict, List, Optional from src.server.models.config import AppConfig, ConfigUpdate, ValidationResult class ConfigServiceError(Exception): """Base exception for configuration service errors.""" class ConfigNotFoundError(ConfigServiceError): """Raised when configuration file is not found.""" class ConfigValidationError(ConfigServiceError): """Raised when configuration validation fails.""" class ConfigBackupError(ConfigServiceError): """Raised when backup operations fail.""" class ConfigService: """Service for managing application configuration persistence. Handles loading, saving, validation, backup, and version management of configuration files. Uses JSON format for human-readable and version-control friendly storage. """ # Current configuration schema version CONFIG_VERSION = "1.0.0" def __init__( self, config_path: Path = Path("data/config.json"), backup_dir: Path = Path("data/config_backups"), max_backups: int = 10 ): """Initialize configuration service. Args: config_path: Path to main configuration file backup_dir: Directory for storing configuration backups max_backups: Maximum number of backups to keep """ self.config_path = config_path self.backup_dir = backup_dir self.max_backups = max_backups # Ensure directories exist self.config_path.parent.mkdir(parents=True, exist_ok=True) self.backup_dir.mkdir(parents=True, exist_ok=True) def load_config(self) -> AppConfig: """Load configuration from file. Returns: AppConfig: Loaded configuration Raises: ConfigNotFoundError: If config file doesn't exist ConfigValidationError: If config validation fails """ if not self.config_path.exists(): # Create default configuration default_config = self._create_default_config() self.save_config(default_config) return default_config try: with open(self.config_path, "r", encoding="utf-8") as f: data = json.load(f) # Remove version key before constructing AppConfig data.pop("version", None) config = AppConfig(**data) # Validate configuration validation = config.validate_config() if not validation.valid: errors = ', '.join(validation.errors or []) raise ConfigValidationError( f"Invalid configuration: {errors}" ) return config except json.JSONDecodeError as e: raise ConfigValidationError( f"Invalid JSON in config file: {e}" ) from e except Exception as e: if isinstance(e, ConfigServiceError): raise raise ConfigValidationError( f"Failed to load config: {e}" ) from e def save_config( self, config: AppConfig, create_backup: bool = True ) -> None: """Save configuration to file. Args: config: Configuration to save create_backup: Whether to create backup before saving Raises: ConfigValidationError: If config validation fails """ # Validate before saving validation = config.validate_config() if not validation.valid: errors = ', '.join(validation.errors or []) raise ConfigValidationError( f"Cannot save invalid configuration: {errors}" ) # Create backup if requested and file exists if create_backup and self.config_path.exists(): try: self.create_backup() except ConfigBackupError as e: # Log but don't fail save operation print(f"Warning: Failed to create backup: {e}") # Save configuration with version data = config.model_dump() data["version"] = self.CONFIG_VERSION # Write to temporary file first for atomic operation temp_path = self.config_path.with_suffix(".tmp") try: with open(temp_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) # Atomic replace temp_path.replace(self.config_path) except Exception as e: # Clean up temp file on error if temp_path.exists(): temp_path.unlink() raise ConfigServiceError(f"Failed to save config: {e}") from e def update_config(self, update: ConfigUpdate) -> AppConfig: """Update configuration with partial changes. Args: update: Configuration update to apply Returns: AppConfig: Updated configuration """ current = self.load_config() updated = update.apply_to(current) self.save_config(updated) return updated def validate_config(self, config: AppConfig) -> ValidationResult: """Validate configuration without saving. Args: config: Configuration to validate Returns: ValidationResult: Validation result with errors if any """ return config.validate_config() def create_backup(self, name: Optional[str] = None) -> Path: """Create backup of current configuration. Args: name: Optional custom backup name (timestamp used if not provided) Returns: Path: Path to created backup file Raises: ConfigBackupError: If backup creation fails """ if not self.config_path.exists(): raise ConfigBackupError("Cannot backup non-existent config file") # Generate backup filename if name is None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") name = f"config_backup_{timestamp}.json" elif not name.endswith(".json"): name = f"{name}.json" backup_path = self.backup_dir / name try: shutil.copy2(self.config_path, backup_path) # Clean up old backups self._cleanup_old_backups() return backup_path except Exception as e: raise ConfigBackupError(f"Failed to create backup: {e}") from e def restore_backup(self, backup_name: str) -> AppConfig: """Restore configuration from backup. Args: backup_name: Name of backup file to restore Returns: AppConfig: Restored configuration Raises: ConfigBackupError: If restore fails """ backup_path = self.backup_dir / backup_name if not backup_path.exists(): raise ConfigBackupError(f"Backup not found: {backup_name}") try: # Create backup of current config before restoring if self.config_path.exists(): self.create_backup("pre_restore") # Restore backup shutil.copy2(backup_path, self.config_path) # Load and validate restored config return self.load_config() except Exception as e: raise ConfigBackupError( f"Failed to restore backup: {e}" ) from e def list_backups(self) -> List[Dict[str, object]]: """List available configuration backups. Returns: List of backup metadata dictionaries with name, size, and created timestamp """ backups: List[Dict[str, object]] = [] if not self.backup_dir.exists(): return backups for backup_file in sorted( self.backup_dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True ): stat = backup_file.stat() created_timestamp = datetime.fromtimestamp(stat.st_mtime) backups.append({ "name": backup_file.name, "size_bytes": stat.st_size, "created_at": created_timestamp.isoformat(), }) return backups def delete_backup(self, backup_name: str) -> None: """Delete a configuration backup. Args: backup_name: Name of backup file to delete Raises: ConfigBackupError: If deletion fails """ backup_path = self.backup_dir / backup_name if not backup_path.exists(): raise ConfigBackupError(f"Backup not found: {backup_name}") try: backup_path.unlink() except OSError as e: raise ConfigBackupError(f"Failed to delete backup: {e}") from e def _create_default_config(self) -> AppConfig: """Create default configuration. Returns: AppConfig: Default configuration """ return AppConfig() def _cleanup_old_backups(self) -> None: """Remove old backups exceeding max_backups limit.""" if not self.backup_dir.exists(): return # Get all backups sorted by modification time (oldest first) backups = sorted( self.backup_dir.glob("*.json"), key=lambda p: p.stat().st_mtime ) # Remove oldest backups if limit exceeded while len(backups) > self.max_backups: oldest = backups.pop(0) try: oldest.unlink() except (OSError, IOError): # Ignore errors during cleanup continue # Singleton instance _config_service: Optional[ConfigService] = None def get_config_service() -> ConfigService: """Get singleton ConfigService instance. Returns: ConfigService: Singleton instance """ global _config_service if _config_service is None: _config_service = ConfigService() return _config_service