Lukas
3cb644add4
- Fix return type annotation in SetupRedirectMiddleware.dispatch() to use Response instead of RedirectResponse - Replace broad 'except Exception' with specific exception types (FileNotFoundError, ValueError, OSError, etc.) - Rename AppConfig.validate() to validate_config() to avoid shadowing BaseModel.validate() - Fix ValidationResult.errors field to use List[str] with default_factory - Add pylint disable comments for intentional broad exception catches during shutdown - Rename lifespan parameter to _application to indicate unused variable - Update all callers to use new validate_config() method name
342 lines
11 KiB
Python
342 lines
11 KiB
Python
"""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
|