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
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
from typing import Dict, List, Optional
|
|
|
|
from pydantic import BaseModel, Field, ValidationError, field_validator
|
|
|
|
|
|
class SchedulerConfig(BaseModel):
|
|
"""Scheduler related configuration."""
|
|
|
|
enabled: bool = Field(
|
|
default=True, description="Whether the scheduler is enabled"
|
|
)
|
|
interval_minutes: int = Field(
|
|
default=60, ge=1, description="Scheduler interval in minutes"
|
|
)
|
|
|
|
|
|
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):
|
|
"""Logging configuration with basic validation for level."""
|
|
|
|
level: str = Field(
|
|
default="INFO", description="Logging level"
|
|
)
|
|
file: Optional[str] = Field(
|
|
default=None, description="Optional file path for log output"
|
|
)
|
|
max_bytes: Optional[int] = Field(
|
|
default=None, ge=0, description="Max bytes per log file for rotation"
|
|
)
|
|
backup_count: Optional[int] = Field(
|
|
default=3, ge=0, description="Number of rotated log files to keep"
|
|
)
|
|
|
|
@field_validator("level")
|
|
@classmethod
|
|
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):
|
|
"""Result of a configuration validation attempt."""
|
|
|
|
valid: bool = Field(..., description="Whether the configuration is valid")
|
|
errors: List[str] = Field(
|
|
default_factory=lambda: [],
|
|
description="List of validation error messages"
|
|
)
|
|
|
|
|
|
class AppConfig(BaseModel):
|
|
"""Top-level application configuration model used by the web layer.
|
|
|
|
This model intentionally keeps things small and serializable to JSON.
|
|
"""
|
|
|
|
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_config(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
|
|
backup_data = self.model_dump().get("backup", {})
|
|
if backup_data.get("enabled") and not backup_data.get("path"):
|
|
errors.append(
|
|
"backup.path must be set when backups.enabled is true"
|
|
)
|
|
|
|
return ValidationResult(valid=(len(errors) == 0), errors=errors)
|
|
|
|
|
|
class ConfigUpdate(BaseModel):
|
|
scheduler: Optional[SchedulerConfig] = None
|
|
logging: Optional[LoggingConfig] = None
|
|
backup: Optional[BackupConfig] = None
|
|
other: Optional[Dict[str, object]] = None
|
|
|
|
def apply_to(self, current: AppConfig) -> AppConfig:
|
|
"""Return a new AppConfig with updates applied to the current config.
|
|
|
|
Performs a shallow merge for `other`.
|
|
"""
|
|
data = current.model_dump()
|
|
if self.scheduler is not None:
|
|
data["scheduler"] = self.scheduler.model_dump()
|
|
if self.logging is not None:
|
|
data["logging"] = self.logging.model_dump()
|
|
if self.backup is not None:
|
|
data["backup"] = self.backup.model_dump()
|
|
if self.other is not None:
|
|
merged = dict(current.other or {})
|
|
merged.update(self.other)
|
|
data["other"] = merged
|
|
return AppConfig(**data)
|