api/logger.py

"""Synchronous prediction logger.

Writes one row to the configured predictions table for every ``/predict``
call (success or failure). Insert failures are logged but never propagated:
the API client has already received its prediction by the time we log, so
losing observability is preferable to losing the response.
"""

from __future__ import annotations

import logging
import math
import time
from typing import Any

import pandas as pd
from sqlalchemy import MetaData, Table, insert, update

from api import db, settings
from database.models import build_predictions_log_table

logger = logging.getLogger(__name__)


def _jsonify_features(features: pd.DataFrame) -> dict[str, Any]:
    """Convert a single-row features DataFrame to a JSON-safe dict.

    NaN/Inf are not valid JSON — they are coerced to None so PostgreSQL
    receives ``null``. The 768 features mix floats, ints, and pandas
    extension types; ``.to_dict()`` plus this scrub handles all of them.
    """
    record = features.iloc[0].to_dict()
    cleaned: dict[str, Any] = {}
    for key, value in record.items():
        if value is None:
            cleaned[key] = None
        elif isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
            cleaned[key] = None
        else:
            try:
                cleaned[key] = value.item()  # numpy scalar -> Python scalar
            except AttributeError:
                cleaned[key] = value
    return cleaned


def _jsonify_raw_input(raw_input: dict[str, Any]) -> dict[str, Any]:
    """Same NaN/Inf scrub for the raw Pydantic dump."""
    cleaned: dict[str, Any] = {}
    for key, value in raw_input.items():
        if isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
            cleaned[key] = None
        else:
            cleaned[key] = value
    return cleaned


def _get_table() -> Table | None:
    """Resolve the active Table bound to the configured engine."""
    engine = db.get_engine()
    if engine is None:
        return None
    metadata = MetaData()
    return build_predictions_log_table(settings.PREDICTIONS_TABLE, metadata)


def log_prediction(
    *,
    sk_id_curr: int,
    client_known: bool,
    raw_input: dict[str, Any],
    features: pd.DataFrame | None,
    probability_default: float | None,
    decision: str | None,
    threshold: float,
    model_version: str,
    latency_ms: int,
    feature_assembly_ms: int | None = None,
    inference_ms: int | None = None,
    inference_cpu_ms: int | None = None,
    plumbing_ms: int | None = None,
    status_code: int = 200,
    error_message: str | None = None,
    top_shap: dict[str, Any] | None = None,
) -> None:
    """Insert one prediction record. Best-effort: never raises.

    Measures the INSERT wall-clock itself as ``db_log_ms`` and writes it on
    the same row — so the full server-side budget can be reconstructed as
    ``latency_ms + db_log_ms``.
    """
    table = _get_table()
    if table is None:
        return  # logging disabled, nothing to do

    payload = {
        "sk_id_curr": sk_id_curr,
        "client_known": client_known,
        "latency_ms": latency_ms,
        "feature_assembly_ms": feature_assembly_ms,
        "inference_ms": inference_ms,
        "inference_cpu_ms": inference_cpu_ms,
        "plumbing_ms": plumbing_ms,
        # Filled in after the INSERT completes below; included in payload now
        # for column ordering but overwritten before send.
        "db_log_ms": None,
        "status_code": status_code,
        "error_message": error_message,
        "raw_input": _jsonify_raw_input(raw_input),
        "features": _jsonify_features(features) if features is not None else {},
        "probability_default": probability_default,
        "decision": decision or "ERROR",
        "threshold": threshold,
        "model_version": model_version,
        "top_shap": top_shap,
    }

    engine = db.get_engine()
    if engine is None:
        # _get_table() returned a Table only because the engine was set when
        # it was called. A race / re-entrancy between the two calls would
        # land here; rather than rely on an ``assert`` (stripped under -O),
        # fail closed: skip the insert silently. The client already has its
        # prediction.
        return
    # Self-describing rows: we want each prediction to carry the wall-clock
    # cost of its own INSERT, so the dashboard can decompose latency into
    # handler + DB log without external joins. The pattern is INSERT with
    # RETURNING id, measure, then UPDATE that exact id. Cost: 2 round-trips
    # per logged prediction — acceptable for monitoring at low QPS, and a
    # documented trade-off in the étape 4 report. If the UPDATE fails after
    # the INSERT succeeded, the row remains in the DB with ``db_log_ms``
    # NULL — acceptable, dashboard PERCENTILE_CONT ignores NULLs natively.
    try:
        t_insert = time.perf_counter()
        with engine.begin() as conn:
            inserted_id = conn.execute(
                insert(table).values(**payload).returning(table.c.id)
            ).scalar_one()
        db_log_ms = round((time.perf_counter() - t_insert) * 1000.0)
        with engine.begin() as conn:
            conn.execute(
                update(table)
                .where(table.c.id == inserted_id)
                .values(db_log_ms=db_log_ms)
            )
    except Exception as exc:  # noqa: BLE001 — best-effort logger
        logger.warning(
            "Failed to log prediction for sk_id=%s: %s", sk_id_curr, exc
        )