pipu-music/services/recommender/pipu_rec/api/app.py

"""FastAPI-приложение внутреннего API рекомендаций."""

import hmac
from collections.abc import Awaitable, Callable
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, status
from fastapi.responses import JSONResponse

from pipu_rec.settings import Settings

Check = Callable[[], Awaitable[None]]


def create_app(settings: Settings, checks: dict[str, Check] | None = None) -> FastAPI:
    app = FastAPI(title="pipu-recommender", version=settings.version, docs_url=None, redoc_url=None)
    checks = checks or {}

    def require_token(authorization: Annotated[str, Header()] = "") -> None:
        scheme, _, token = authorization.partition(" ")
        expected = settings.internal_token
        if scheme != "Bearer" or not expected or not hmac.compare_digest(token, expected):
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="auth.invalid_internal_token")

    @app.get("/healthz")
    async def healthz() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/readyz")
    async def readyz() -> JSONResponse:
        results: dict[str, str] = {}
        for name, check in checks.items():
            try:
                await check()
                results[name] = "ok"
            except Exception as exc:
                results[name] = str(exc) or type(exc).__name__
        healthy = all(v == "ok" for v in results.values())
        return JSONResponse(
            {"status": "ok" if healthy else "degraded", "checks": results},
            status_code=status.HTTP_200_OK if healthy else status.HTTP_503_SERVICE_UNAVAILABLE,
        )

    @app.get("/internal/status", dependencies=[Depends(require_token)])
    async def internal_status() -> dict[str, object]:
        return {"version": settings.version, "models": [], "ranker": "heuristic-v1"}

    return app