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.
52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
# ============================================================
|
|
# helpers.py — Conversion des résultats pyodbc et journal d'audit
|
|
# ============================================================
|
|
|
|
from fastapi import HTTPException, status
|
|
|
|
from domain import MONITO_TABLES, AuditAction
|
|
|
|
|
|
def rows_to_list(cursor, rows) -> list[dict]:
|
|
"""Convertit les lignes pyodbc en liste de dicts."""
|
|
cols = [col[0] for col in cursor.description]
|
|
return [dict(zip(cols, r)) for r in rows]
|
|
|
|
|
|
def row_to_dict(cursor, row) -> dict:
|
|
"""Convertit une ligne pyodbc en dict."""
|
|
cols = [col[0] for col in cursor.description]
|
|
return dict(zip(cols, row))
|
|
|
|
|
|
def get_table_name(id_monito: int) -> str:
|
|
"""Retourne le nom de la table dédiée. Lève 404 si inconnu."""
|
|
table = MONITO_TABLES.get(id_monito)
|
|
if not table:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Monitoring {id_monito} introuvable. IDs valides : {list(MONITO_TABLES.keys())}"
|
|
)
|
|
return table
|
|
|
|
|
|
def serialize_row(row: dict) -> dict:
|
|
"""Convertit les types non-JSON (date, Decimal) en types sérialisables."""
|
|
result = {}
|
|
for k, v in row.items():
|
|
if v is None:
|
|
result[k] = None
|
|
elif hasattr(v, 'isoformat'):
|
|
result[k] = v.isoformat()
|
|
else:
|
|
result[k] = v
|
|
return result
|
|
|
|
|
|
def journaliser(cursor, acteur: dict, action: AuditAction, detail: str) -> None:
|
|
"""Trace une action d'administration dans JOURNAL_AUDIT."""
|
|
cursor.execute(
|
|
"INSERT INTO JOURNAL_AUDIT (id_user, username, action, detail) VALUES (?, ?, ?, ?)",
|
|
acteur["id_user"], acteur["username"], action.value, detail
|
|
)
|