src/main.py - master - maruga/mlops

src/main.py
118 строк4 KB
s546371
modified: .gitlab-ci.yml
29 мар 2026, 21:35
29 мар 2026, 21:3547c4e99
import logging
from typing import Annotated, Dict, Union

from fastapi import Depends, FastAPI
from fastapi.responses import JSONResponse, Response
from prometheus_client import (
    CONTENT_TYPE_LATEST,
    Counter,
    Gauge,
    Histogram,
    generate_latest,
)

from src.config import DataMap
from src.loaded_models import item_features, ranker
from src.schema import ErrorResponse, PredictPurchaseRequest, PredictPurchaseResponse

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)

APP_VERSION = "0.0.1"

app = FastAPI(
    title="Retail Recommendation API",
    version=APP_VERSION,
    description="API для выдачи рекомендаций и вероятностей покупки.",
)

# --- Prometheus метрики ---
REQUEST_COUNT = Counter(
    "app_requests_total",
    "Общее количество запросов к приложению",
    ["endpoint", "http_status"],
)
REQUEST_LATENCY = Histogram(
    "app_request_latency_seconds", "Задержка обработки запроса в секундах", ["endpoint"]
)
IN_PROGRESS = Gauge(
    "app_inprogress_requests", "Количество одновременно обрабатываемых запросов"
)
# --- Endpoints ---


@app.get("/health", tags=["system"])
def health_check() -> Dict[str, Union[str, Dict[str, int]]]:
    """
    Проверка состояния сервиса и доступности моделей.
    """

    ok = {
        "ranker_loaded": ranker is not None,
        "items_in_feature_store": len(item_features),
    }
    logger.info(f"Health check: {ok}")
    return {"status": "ok", "details": ok}


@app.get("/version", tags=["system"])
def version() -> Dict[str, str]:
    """
    Возвращает текущую версию сервиса.
    """
    return {"version": APP_VERSION, "service": "retail_recsys"}


@app.get("/", tags=["system"])
def root() -> Dict[str, str]:
    return {"message": "Retail recommender API running", "version": APP_VERSION}


@app.get(
    "/predict_purchase",
    response_model=PredictPurchaseResponse,
    responses={
        200: {
            "model": PredictPurchaseResponse,
            "description": "Успешный ответ с вероятностью покупки",
        },
        404: {"model": ErrorResponse, "description": "Товар не найден"},
    },
)
def predict_purchase(
    params: Annotated[PredictPurchaseRequest, Depends()],
) -> PredictPurchaseResponse | JSONResponse:
    """Предсказываем вероятность покупки товара в конкретный день недели и время

    - item_id: номер товара в базе данных
    - hour: час текущий
    - weekday: день недели текущий (Понедельник=0)

    Вернётся объект с item_id и purchase_probability или объект с полем error
    """
    row = item_features[item_features["itemid"] == params.item_id].copy()
    if row.empty:
        return JSONResponse(
            content=ErrorResponse(error="item not found").model_dump(), status_code=404
        )

    row[DataMap.hour] = params.hour
    row[DataMap.weekday] = params.weekday
    x = row[DataMap.features].astype(float)
    prob = float(ranker.predict_proba(x)[0][1])
    return PredictPurchaseResponse(item_id=params.item_id, purchase_probability=prob)


@app.get("/metrics")
def metrics() -> Response:
    """Endpoint для Prometheus"""
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("src.main:app", host="0.0.0.0", port=32000, reload=True)