"""Pydantic models for backend services (Semgrep: BaseModel lives in models.py).""" from __future__ import annotations import json from pathlib import Path from typing import Literal from pydantic import BaseModel, Field, field_validator RecsysSource = Literal["graph", "ots", "hybrid"] class RecsysSettings(BaseModel): """Production and tests configure recommender behavior explicitly or via optional JSON.""" source: RecsysSource = Field(default="graph", description="graph | ots | hybrid") candidates_path: Path | None = Field( default=None, description=( "Override path to OTS candidate JSON; default is " "/recsys/candidates.json" ), ) @field_validator("source", mode="before") @classmethod def normalize_source_aliases(cls, value: str | int | float | bool | None) -> str: if value is None: return "graph" raw = str(value).strip().lower() if raw in {"", "graph", "current"}: return "graph" if raw in {"ots", "cornac", "recbole"}: return "ots" if raw == "hybrid": return "hybrid" return "graph" def load_recsys_settings_from_store( store_path: Path, override: RecsysSettings | None, ) -> RecsysSettings: """Load typed settings from disk (no process environment).""" if override is not None: return override.model_copy(deep=True) config_path = store_path.parent / "recsys" / "settings.json" if not config_path.is_file(): base = RecsysSettings() else: text = config_path.read_text(encoding="utf-8") data = json.loads(text) base = RecsysSettings.model_validate(data) if base.candidates_path is not None and not base.candidates_path.is_absolute(): rel = base.candidates_path base = base.model_copy( update={ "candidates_path": config_path.parent / rel, }, ) return base