kink-discovery / backend /models.py
Perplexed7675's picture
Sync from kink_cli (Docker Space)
6ff91d6 verified
Raw
History Blame Contribute Delete
2.03 kB
"""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 "
"<store_dir>/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