feat(config): add Pydantic AppConfig, BackupConfig, LoggingConfig; update tests and infra notes

This commit is contained in:
Lukas 2025-10-14 21:43:48 +02:00
parent 4aa7adba3a
commit 52b96da8dc
3 changed files with 101 additions and 39 deletions

View File

@ -45,14 +45,6 @@ The tasks should be completed in the following order to ensure proper dependenci
### 3. Configuration Management ### 3. Configuration Management
#### [] Create configuration service
- []Create `src/server/services/config_service.py`
- []Implement configuration loading/saving
- []Add configuration validation
- []Include backup/restore functionality
- []Add scheduler configuration management
#### [] Implement configuration API endpoints #### [] Implement configuration API endpoints
- []Create `src/server/api/config.py` - []Create `src/server/api/config.py`

View File

@ -1,9 +1,11 @@
from typing import List, Optional from typing import Dict, List, Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field, ValidationError, validator
class SchedulerConfig(BaseModel): class SchedulerConfig(BaseModel):
"""Scheduler related configuration."""
enabled: bool = Field( enabled: bool = Field(
default=True, description="Whether the scheduler is enabled" default=True, description="Whether the scheduler is enabled"
) )
@ -11,12 +13,27 @@ class SchedulerConfig(BaseModel):
default=60, ge=1, description="Scheduler interval in minutes" default=60, ge=1, description="Scheduler interval in minutes"
) )
# ge=1 on the Field enforces a positive interval; no custom validator
# is required. class BackupConfig(BaseModel):
"""Configuration for automatic backups of application data."""
enabled: bool = Field(
default=False, description="Whether backups are enabled"
)
path: Optional[str] = Field(
default="data/backups", description="Path to store backups"
)
keep_days: int = Field(
default=30, ge=0, description="How many days to keep backups"
)
class LoggingConfig(BaseModel): class LoggingConfig(BaseModel):
level: str = Field(default="INFO", description="Logging level") """Logging configuration with basic validation for level."""
level: str = Field(
default="INFO", description="Logging level"
)
file: Optional[str] = Field( file: Optional[str] = Field(
default=None, description="Optional file path for log output" default=None, description="Optional file path for log output"
) )
@ -27,40 +44,87 @@ class LoggingConfig(BaseModel):
default=3, ge=0, description="Number of rotated log files to keep" default=3, ge=0, description="Number of rotated log files to keep"
) )
@validator("level")
def validate_level(cls, v: str) -> str:
allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
lvl = (v or "").upper()
if lvl not in allowed:
raise ValueError(f"invalid logging level: {v}")
return lvl
class ValidationResult(BaseModel): class ValidationResult(BaseModel):
"""Result of a configuration validation attempt."""
valid: bool = Field(..., description="Whether the configuration is valid") valid: bool = Field(..., description="Whether the configuration is valid")
errors: Optional[List[str]] = Field( errors: Optional[List[str]] = Field(
default_factory=list, description="List of validation error messages" default_factory=list, description="List of validation error messages"
) )
class ConfigResponse(BaseModel): class AppConfig(BaseModel):
scheduler: SchedulerConfig """Top-level application configuration model used by the web layer.
logging: LoggingConfig
other: Optional[dict] = Field( This model intentionally keeps things small and serializable to JSON.
default_factory=dict, description="Other arbitrary config values" """
name: str = Field(default="Aniworld", description="Application name")
data_dir: str = Field(default="data", description="Base data directory")
scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig)
logging: LoggingConfig = Field(default_factory=LoggingConfig)
backup: BackupConfig = Field(default_factory=BackupConfig)
other: Dict[str, object] = Field(
default_factory=dict, description="Arbitrary other settings"
) )
def validate(self) -> ValidationResult:
"""Perform light-weight validation and return a ValidationResult.
This method intentionally avoids performing IO (no filesystem checks)
so it remains fast and side-effect free for unit tests and API use.
"""
errors: List[str] = []
# Pydantic field validators already run on construction; re-run a
# quick check for common constraints and collect messages.
try:
# Reconstruct to ensure nested validators are executed
AppConfig(**self.model_dump())
except ValidationError as exc:
for e in exc.errors():
loc = ".".join(str(x) for x in e.get("loc", []))
msg = f"{loc}: {e.get('msg')}"
errors.append(msg)
# backup.path must be set when backups are enabled
if self.backup.enabled and (not self.backup.path):
errors.append(
"backup.path must be set when backups.enabled is true"
)
return ValidationResult(valid=(len(errors) == 0), errors=errors)
class ConfigUpdate(BaseModel): class ConfigUpdate(BaseModel):
scheduler: Optional[SchedulerConfig] = None scheduler: Optional[SchedulerConfig] = None
logging: Optional[LoggingConfig] = None logging: Optional[LoggingConfig] = None
other: Optional[dict] = None backup: Optional[BackupConfig] = None
other: Optional[Dict[str, object]] = None
def apply_to(self, current: ConfigResponse) -> ConfigResponse: def apply_to(self, current: AppConfig) -> AppConfig:
"""Return a new ConfigResponse with updates applied to the current """Return a new AppConfig with updates applied to the current config.
config.
Performs a shallow merge for `other`.
""" """
# Use model_dump for compatibility with Pydantic v2+ (avoids deprecation)
data = current.model_dump() data = current.model_dump()
if self.scheduler is not None: if self.scheduler is not None:
data["scheduler"] = self.scheduler.model_dump() data["scheduler"] = self.scheduler.model_dump()
if self.logging is not None: if self.logging is not None:
data["logging"] = self.logging.model_dump() data["logging"] = self.logging.model_dump()
if self.backup is not None:
data["backup"] = self.backup.model_dump()
if self.other is not None: if self.other is not None:
# shallow merge
merged = dict(current.other or {}) merged = dict(current.other or {})
merged.update(self.other) merged.update(self.other)
data["other"] = merged data["other"] = merged
return ConfigResponse(**data) return AppConfig(**data)

View File

@ -1,7 +1,7 @@
import pytest import pytest
from src.server.models.config import ( from src.server.models.config import (
ConfigResponse, AppConfig,
ConfigUpdate, ConfigUpdate,
LoggingConfig, LoggingConfig,
SchedulerConfig, SchedulerConfig,
@ -25,25 +25,31 @@ def test_logging_config_defaults_and_values():
assert log.backup_count == 3 assert log.backup_count == 3
def test_config_update_apply_to(): def test_appconfig_and_config_update_apply_to():
base = ConfigResponse( base = AppConfig()
scheduler=SchedulerConfig(),
logging=LoggingConfig(),
other={"a": 1},
)
upd = ConfigUpdate(scheduler=SchedulerConfig(enabled=False, interval_minutes=30)) upd = ConfigUpdate(
scheduler=SchedulerConfig(enabled=False, interval_minutes=30)
)
new = upd.apply_to(base) new = upd.apply_to(base)
assert isinstance(new, AppConfig)
assert new.scheduler.enabled is False assert new.scheduler.enabled is False
assert new.scheduler.interval_minutes == 30 assert new.scheduler.interval_minutes == 30
upd2 = ConfigUpdate(other={"b": 2}) upd2 = ConfigUpdate(other={"b": 2})
new2 = upd2.apply_to(base) new2 = upd2.apply_to(base)
assert new2.other["a"] == 1 assert new2.other.get("b") == 2
assert new2.other["b"] == 2
def test_validation_result_model(): def test_backup_and_validation():
vr = ValidationResult(valid=True) cfg = AppConfig()
assert vr.valid is True # default backups disabled -> valid
assert isinstance(vr.errors, list) res: ValidationResult = cfg.validate()
assert res.valid is True
# enable backups but leave path empty -> invalid
cfg.backup.enabled = True
cfg.backup.path = ""
res2 = cfg.validate()
assert res2.valid is False
assert any("backup.path" in e for e in res2.errors)