api/monitoring/metrics.py

from src.api.monitoring.db import get_connection

def _apply_monitoring_filters(query, params, endpoint=None, time_window=None):
    if endpoint and endpoint != "all":
        query += " AND endpoint = %s"
        params.append(endpoint.replace("/", ""))

    if time_window == "24h":
        query += " AND timestamp >= NOW() - INTERVAL '24 hours'"

    elif time_window == "7j":
        query += " AND timestamp >= NOW() - INTERVAL '7 days'"

    elif time_window == "30j":
        query += " AND timestamp >= NOW() - INTERVAL '30 days'"

    return query, params


def get_monitoring_summary():
    """Récupère un résumé des métriques de monitoring à partir de la base de données 
    en exécutant une requête SQL pour calculer :
     - le nombre total de requêtes
     - la latence moyenne
     - le nombre de succès
     - le nombre d'erreurs
     - le taux de succès
    à partir de la table api_requests, puis retourne ces métriques sous forme de dictionnaire.
    Returns:
        dict: Un dictionnaire contenant les métriques de monitoring, y compris le nombre total de requêtes, la latence moyenne, le nombre de succès, le nombre d'erreurs et le taux de succès.
    """
    query = """
    SELECT
        COUNT(*) AS total_requests,
        ROUND(AVG(latency_ms)::numeric, 2) AS avg_latency,
        SUM(CASE WHEN success = true THEN 1 ELSE 0 END) AS success_count,
        SUM(CASE WHEN success = false THEN 1 ELSE 0 END) AS error_count
    FROM api_requests;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query)
            row = cursor.fetchone()

    total_requests, avg_latency, success_count, error_count = row

    success_rate = 0

    if total_requests and total_requests > 0:
        success_rate = round(
            (success_count / total_requests) * 100,
            2
        )

    return {
        "total_requests": total_requests,
        "avg_latency_ms": avg_latency,
        "success_count": success_count,
        "error_count": error_count,
        "success_rate": success_rate
    }

def get_endpoint_usage(endpoint=None, time_window=None):
    query = """
    SELECT
        endpoint,
        COUNT(*) AS total
    FROM api_requests
    WHERE 1=1
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    GROUP BY endpoint
    ORDER BY total DESC;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "endpoint": row[0],
            "total": row[1],
        }
        for row in rows
    ]

def get_recent_requests(limit=20, endpoint=None, time_window=None):
    query = """
    SELECT
        timestamp,
        endpoint,
        status_code,
        success,
        latency_ms,
        model_name
    FROM api_requests
    WHERE 1=1
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    ORDER BY timestamp DESC
    LIMIT %s;
    """

    params.append(limit)

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "timestamp": row[0],
            "endpoint": row[1],
            "status_code": row[2],
            "success": row[3],
            "latency_ms": row[4],
            "model_name": row[5],
        }
        for row in rows
    ]

def get_latency_by_endpoint(endpoint=None, time_window=None):
    query = """
    SELECT
        endpoint,
        COUNT(*) AS total_requests,
        ROUND(AVG(latency_ms)::numeric, 2) AS avg_latency_ms,
        ROUND(MIN(latency_ms)::numeric, 2) AS min_latency_ms,
        ROUND(MAX(latency_ms)::numeric, 2) AS max_latency_ms
    FROM api_requests
    WHERE 1=1
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    GROUP BY endpoint
    ORDER BY AVG(latency_ms) DESC;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "endpoint": row[0],
            "total_requests": row[1],
            "avg_latency_ms": float(row[2]) if row[2] is not None else 0,
            "min_latency_ms": float(row[3]) if row[3] is not None else 0,
            "max_latency_ms": float(row[4]) if row[4] is not None else 0,
        }
        for row in rows
    ]

def get_inference_step_metrics():
    """
    Récupère les métriques de latence
    par étape d'inférence.
    """

    query = """
    SELECT
        step_name,
        COUNT(*) AS total_runs,
        ROUND(AVG(duration_ms)::numeric, 2) AS avg_duration_ms,
        ROUND(MIN(duration_ms)::numeric, 2) AS min_duration_ms,
        ROUND(MAX(duration_ms)::numeric, 2) AS max_duration_ms
    FROM inference_steps
    GROUP BY step_name
    ORDER BY avg_duration_ms DESC;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query)
            rows = cursor.fetchall()

    return [
        {
            "step_name": row[0],
            "total_runs": row[1],
            "avg_duration_ms": float(row[2]) if row[2] else 0,
            "min_duration_ms": float(row[3]) if row[3] else 0,
            "max_duration_ms": float(row[4]) if row[4] else 0,
        }
        for row in rows
    ]

def get_recent_errors(limit=20, endpoint=None, time_window=None):
    query = """
    SELECT
        timestamp,
        endpoint,
        status_code,
        error_message,
        latency_ms
    FROM api_requests
    WHERE success = false
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    ORDER BY timestamp DESC
    LIMIT %s;
    """

    params.append(limit)

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "timestamp": row[0],
            "endpoint": row[1],
            "status_code": row[2],
            "error_message": row[3],
            "latency_ms": row[4],
        }
        for row in rows
    ]

def get_top_recommended_crops(limit=10):
    query = """
    SELECT
        recommended_crop,
        COUNT(*) AS total
    FROM api_predictions
    WHERE prediction_type = 'recommend'
      AND recommended_crop IS NOT NULL
    GROUP BY recommended_crop
    ORDER BY total DESC
    LIMIT %s;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, (limit,))
            rows = cursor.fetchall()

    return [
        {"recommended_crop": row[0], "total": row[1]}
        for row in rows
    ]

def get_latency_by_endpoint(endpoint=None, time_window=None):
    query = """
    SELECT
        endpoint,
        COUNT(*) AS total_requests,
        ROUND(AVG(latency_ms)::numeric, 2) AS avg_latency_ms,
        ROUND(MIN(latency_ms)::numeric, 2) AS min_latency_ms,
        ROUND(MAX(latency_ms)::numeric, 2) AS max_latency_ms
    FROM api_requests
    WHERE 1=1
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    GROUP BY endpoint
    ORDER BY AVG(latency_ms) DESC;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "endpoint": row[0],
            "total_requests": row[1],
            "avg_latency_ms": float(row[2]) if row[2] is not None else 0,
            "min_latency_ms": float(row[3]) if row[3] is not None else 0,
            "max_latency_ms": float(row[4]) if row[4] is not None else 0,
        }
        for row in rows
    ]

def get_requests_over_time(endpoint=None, time_window=None):
    query = """
    SELECT
        DATE_TRUNC('hour', timestamp) AS period,
        COUNT(*) AS total_requests
    FROM api_requests
    WHERE 1=1
    """

    params = []

    query, params = _apply_monitoring_filters(
        query=query,
        params=params,
        endpoint=endpoint,
        time_window=time_window,
    )

    query += """
    GROUP BY period
    ORDER BY period ASC;
    """

    with get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute(query, tuple(params))
            rows = cursor.fetchall()

    return [
        {
            "period": row[0],
            "total_requests": row[1],
        }
        for row in rows
    ]