Lukas
51b7f349f8
SchedulerConfig.__init__ maps legacy auto_download/folder_scan keys to the primary auto_download_after_rescan/folder_scan_enabled fields. However, model_dump() was including auto_download=null and folder_scan=null in serialised output. When this was written to config.json and reloaded, those keys were present (albeit null), so the alias mapping was skipped and the primary fields retained default False values instead of the configured True values. Fix: - Override SchedulerConfig.model_dump() to drop None-valued alias fields before returning the serialised dict. - ConfigService.save_config() re-serialises the scheduler field through its overridden model_dump() so the fix applies when writing to disk. Tests added: - test_roundtrip_excludes_none_alias_fields: verifies model_dump omits null auto_download/folder_scan keys. - test_save_and_load_scheduler_flags_roundtrip: end-to-end roundtrip through ConfigService confirms raw JSON and loaded values match. Pre-existing failure in test_core_error_handler.py is unrelated.
351 lines
11 KiB
Python
351 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 logging
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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.1"
|
|
|
|
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
|
|
logger.warning("Failed to create backup: %s", e)
|
|
|
|
# Save configuration with version
|
|
data = config.model_dump()
|
|
data["version"] = self.CONFIG_VERSION
|
|
|
|
# Re-serialize SchedulerConfig through its overridden model_dump so
|
|
# that None legacy alias fields are stripped before writing to disk.
|
|
# Pydantic converts nested models to plain dicts in model_dump() output,
|
|
# so we call the override explicitly on the scheduler field.
|
|
data["scheduler"] = config.scheduler.model_dump()
|
|
|
|
# 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
|