Antho
934e669774
main.py comptait 829 lignes et regroupait 9 domaines fonctionnels. Il ne fait plus que l'assemblage de l'application (configuration, middlewares, montage des routeurs), soit 115 lignes. - routers/ : un module par domaine, chacun déclarant sa propre dépendance d'authentification - domain.py : enums UserRole et AuditAction, mapping MONITO_TABLES ; les rôles étaient jusqu'ici répétés en dur à deux endroits - helpers.py : conversion des lignes pyodbc, écriture du journal - rate_limit.py : limiteur partagé, isolé pour éviter un import circulaire entre main.py et le routeur d'authentification Les codes HTTP littéraux (404, 401, 400, 201) passent aux constantes fastapi.status, comme le faisait déjà auth.py. Les actions du journal d'audit passent en paramètre SQL au lieu d'être concaténées. Aucune route modifiée : la comparaison des specs OpenAPI avant/après confirme que les 26 URL existantes sont identiques. conftest patchait main.get_cursor ; chaque routeur important désormais get_cursor dans son propre espace de noms, la fixture remplace le nom dans tous les modules concernés.
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
# ============================================================
|
|
# routers/dashboard.py — Vue consolidée (VUE_CONSO)
|
|
# ============================================================
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
|
|
from auth import get_current_user
|
|
from config import get_cursor
|
|
from helpers import row_to_dict, rows_to_list
|
|
|
|
router = APIRouter(dependencies=[Depends(get_current_user)], tags=["Dashboard"])
|
|
|
|
|
|
@router.get("")
|
|
def get_dashboard(
|
|
service : Optional[str] = Query(None, description="Filtrer par service"),
|
|
categorie : Optional[str] = Query(None, description="Filtrer par catégorie"),
|
|
):
|
|
"""Vue consolidée — source principale du dashboard."""
|
|
query = (
|
|
"SELECT id_monito, nom_monito, nb_erreurs, service, categorie, bdd_source "
|
|
"FROM VUE_CONSO WHERE 1=1"
|
|
)
|
|
params = []
|
|
|
|
if service:
|
|
query += " AND service = ?"
|
|
params.append(service)
|
|
if categorie:
|
|
query += " AND categorie = ?"
|
|
params.append(categorie)
|
|
|
|
query += " ORDER BY nb_erreurs DESC"
|
|
|
|
with get_cursor() as cursor:
|
|
cursor.execute(query, *params)
|
|
return rows_to_list(cursor, cursor.fetchall())
|
|
|
|
|
|
@router.get("/filtres")
|
|
def get_dashboard_filtres():
|
|
"""
|
|
Valeurs de filtre réellement présentes dans VUE_CONSO.
|
|
|
|
Les référentiels SERVICE / CATEGORIE contiennent des entrées auxquelles
|
|
aucun monitoring actif n'est rattaché (ex. « Business Intelligence »).
|
|
Les proposer dans les menus déroulants du dashboard mène à un écran vide :
|
|
cet endpoint ne renvoie que les valeurs qui ramènent au moins un monitoring.
|
|
|
|
`combinaisons` permet au frontend de restreindre les catégories proposées
|
|
au service sélectionné.
|
|
"""
|
|
with get_cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT DISTINCT service, categorie FROM VUE_CONSO "
|
|
"ORDER BY service, categorie"
|
|
)
|
|
combinaisons = [{"service": r[0], "categorie": r[1]} for r in cursor.fetchall()]
|
|
|
|
return {
|
|
"services" : sorted({c["service"] for c in combinaisons}),
|
|
"categories" : sorted({c["categorie"] for c in combinaisons}),
|
|
"combinaisons" : combinaisons,
|
|
}
|
|
|
|
|
|
@router.get("/summary")
|
|
def get_dashboard_summary():
|
|
"""KPI globaux pour les 4 cartes du dashboard."""
|
|
with get_cursor() as cursor:
|
|
cursor.execute("""
|
|
SELECT
|
|
COUNT(*) AS nb_monitorings,
|
|
SUM(nb_erreurs) AS total_erreurs,
|
|
MAX(nb_erreurs) AS max_erreurs,
|
|
SUM(CASE WHEN nb_erreurs = 0 THEN 1 ELSE 0 END) AS monitorings_ok,
|
|
SUM(CASE WHEN nb_erreurs > 0 THEN 1 ELSE 0 END) AS monitorings_en_erreur
|
|
FROM VUE_CONSO
|
|
""")
|
|
summary = row_to_dict(cursor, cursor.fetchone())
|
|
|
|
cursor.execute(
|
|
"SELECT TOP 1 id_monito, nom_monito, nb_erreurs, service, bdd_source "
|
|
"FROM VUE_CONSO ORDER BY nb_erreurs DESC"
|
|
)
|
|
top = cursor.fetchone()
|
|
summary["monitoring_critique"] = row_to_dict(cursor, top) if top else None
|
|
|
|
return summary
|