feat: Add comprehensive configuration persistence system
- Implemented ConfigService with file-based JSON persistence
- Atomic file writes using temporary files
- Configuration validation with detailed error reporting
- Schema versioning with migration support
- Singleton pattern for global access
- Added backup management functionality
- Automatic backup creation before updates
- Manual backup creation with custom names
- Backup restoration with pre-restore backup
- Backup listing and deletion
- Automatic cleanup of old backups (max 10)
- Updated configuration API endpoints
- GET /api/config - Retrieve configuration
- PUT /api/config - Update with automatic backup
- POST /api/config/validate - Validation without applying
- GET /api/config/backups - List all backups
- POST /api/config/backups - Create manual backup
- POST /api/config/backups/{name}/restore - Restore backup
- DELETE /api/config/backups/{name} - Delete backup
- Comprehensive test coverage
- 27 unit tests for ConfigService (all passing)
- Integration tests for API endpoints
- Tests for validation, persistence, backups, and error handling
- Updated documentation
- Added ConfigService documentation to infrastructure.md
- Marked task as completed in instructions.md
Files changed:
- src/server/services/config_service.py (new)
- src/server/api/config.py (refactored)
- tests/unit/test_config_service.py (new)
- tests/api/test_config_endpoints.py (enhanced)
- infrastructure.md (updated)
- instructions.md (updated)
This commit is contained in:
366
src/server/services/config_service.py
Normal file
366
src/server/services/config_service.py
Normal file
@@ -0,0 +1,366 @@
|
||||
"""Configuration persistence service for managing application settings.
|
||||
|
||||
This service handles:
|
||||
- Loading and saving configuration to JSON files
|
||||
- Configuration validation
|
||||
- Backup and restore functionality
|
||||
- Configuration migration for version updates
|
||||
"""
|
||||
|
||||
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 migration 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)
|
||||
|
||||
# Check if migration is needed
|
||||
file_version = data.get("version", "1.0.0")
|
||||
if file_version != self.CONFIG_VERSION:
|
||||
data = self._migrate_config(data, file_version)
|
||||
|
||||
# Remove version key before constructing AppConfig
|
||||
data.pop("version", None)
|
||||
|
||||
config = AppConfig(**data)
|
||||
|
||||
# Validate configuration
|
||||
validation = config.validate()
|
||||
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()
|
||||
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()
|
||||
|
||||
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
|
||||
|
||||
def _migrate_config(
|
||||
self, data: Dict, from_version: str # noqa: ARG002
|
||||
) -> Dict:
|
||||
"""Migrate configuration from old version to current.
|
||||
|
||||
Args:
|
||||
data: Configuration data to migrate
|
||||
from_version: Version to migrate from (reserved for future use)
|
||||
|
||||
Returns:
|
||||
Dict: Migrated configuration data
|
||||
"""
|
||||
# Currently only one version exists
|
||||
# Future migrations would go here
|
||||
# Example:
|
||||
# if from_version == "1.0.0" and self.CONFIG_VERSION == "2.0.0":
|
||||
# data = self._migrate_1_0_to_2_0(data)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
# 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
|
||||
Reference in New Issue
Block a user