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 ]