api/monitoring/logging.py

from datetime import datetime

from psycopg2.extras import Json

from src.api.monitoring.db import get_connection


def log_api_request(
    endpoint,
    status_code,
    success,
    latency_ms,
    model_name,
    model_version,
    input_data,
    output_data=None,
    error_message=None,
):
    """génère une entrée dans la table api_requests et retourne l'id de la requête
    Args:
        endpoint (str): Le point de terminaison de l'API appelé (ex: "predict", "recommend").
        status_code (int): Le code de statut HTTP de la réponse (ex: 200, 400).
        success (bool): Indique si la requête a été traitée avec succès.    
        latency_ms (float): Le temps de latence total de la requête en millisecondes.
        model_name (str): Le nom du modèle utilisé pour la prédiction.
        model_version (str): La version du modèle utilisé pour la prédiction.
        input_data (dict): Les données d'entrée envoyées à l'API.
        output_data (dict, optional): Les données de sortie retournées par l'API.
        error_message (str, optional): Un message d'erreur si la requête a échoué.
    Returns:
        int: L'id de la requête nouvellement créée dans la table api_requests.
    """
    query = """
    INSERT INTO api_requests (
        timestamp, endpoint, status_code, success, error_message,
        latency_ms, model_name, model_version, input_json, output_json
    )
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    RETURNING id;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                query,
                (
                    datetime.now(),
                    endpoint,
                    status_code,
                    success,
                    error_message,
                    latency_ms,
                    model_name,
                    str(model_version),
                    Json(input_data),
                    Json(output_data),
                ),
            )
            request_id = cursor.fetchone()[0]

    return request_id


def log_inference_step(
    request_id,
    step_name,
    success,
    duration_ms,
    details=None,
    error_message=None,
):
    query = """
    INSERT INTO inference_steps (
        request_id, step_name, success, error_message, duration_ms, details_json
    )
    VALUES (%s, %s, %s, %s, %s, %s);
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                query,
                (
                    request_id,
                    step_name,
                    success,
                    error_message,
                    duration_ms,
                    Json(details),
                ),
            )


def log_api_prediction(
    request_id,
    area,
    prediction_type,
    item=None,
    recommended_crop=None,
    predicted_yield=None,
    n_candidates=None,
):
    query = """
    INSERT INTO api_predictions (
        request_id, area, item, recommended_crop,
        predicted_yield, n_candidates, prediction_type
    )
    VALUES (%s, %s, %s, %s, %s, %s, %s);
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                query,
                (
                    request_id,
                    area,
                    item,
                    recommended_crop,
                    predicted_yield,
                    n_candidates,
                    prediction_type,
                ),
            )


def log_recommendation_details(request_id, ranking):
    query = """
    INSERT INTO api_recommendation_details (
        request_id, rank, item, predicted_yield
    )
    VALUES (%s, %s, %s, %s);
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            for row in ranking:
                cursor.execute(
                    query,
                    (
                        request_id,
                        int(row["rank"]),
                        row["item"],
                        float(row["predicted_yield"]),
                    ),
                )