refactor(api): découpe main.py en routeurs par domaine
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.
This commit is contained in:
@@ -0,0 +1,233 @@
|
||||
# ============================================================
|
||||
# routers/referentiels.py — Services, catégories et contacts
|
||||
#
|
||||
# Lecture ouverte à tout utilisateur authentifié ; création,
|
||||
# modification et suppression réservées aux administrateurs.
|
||||
# ============================================================
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from auth import get_current_user, require_admin
|
||||
from config import get_cursor
|
||||
from domain import AuditAction
|
||||
from helpers import journaliser, row_to_dict, rows_to_list
|
||||
|
||||
router = APIRouter(dependencies=[Depends(get_current_user)], tags=["Référentiels"])
|
||||
|
||||
|
||||
class ServiceBody(BaseModel):
|
||||
nom_service: str = Field(min_length=1, max_length=100)
|
||||
|
||||
|
||||
class CategorieBody(BaseModel):
|
||||
intitule_categorie: str = Field(min_length=1, max_length=100)
|
||||
|
||||
|
||||
class ContactBody(BaseModel):
|
||||
id_service: int
|
||||
intitule_contact: str = Field(min_length=1, max_length=100)
|
||||
nom: str = Field(min_length=1, max_length=100)
|
||||
prenom: str = Field(min_length=1, max_length=100)
|
||||
mail: str = Field(min_length=1, max_length=200)
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Catégories
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@router.get("/categories")
|
||||
def get_categories():
|
||||
"""Toutes les catégories de monitoring."""
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT id_categorie, intitule_categorie "
|
||||
"FROM CATEGORIE ORDER BY intitule_categorie"
|
||||
)
|
||||
return rows_to_list(cursor, cursor.fetchall())
|
||||
|
||||
|
||||
@router.post("/categories", status_code=status.HTTP_201_CREATED)
|
||||
def create_categorie(body: CategorieBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"INSERT INTO CATEGORIE (intitule_categorie) OUTPUT INSERTED.id_categorie VALUES (?)",
|
||||
body.intitule_categorie
|
||||
)
|
||||
id_categorie = cursor.fetchone()[0]
|
||||
journaliser(cursor, admin, AuditAction.CREATE_REFERENTIEL,
|
||||
f"CATEGORIE {id_categorie} : {body.intitule_categorie}")
|
||||
return {"id_categorie": id_categorie, "intitule_categorie": body.intitule_categorie}
|
||||
|
||||
|
||||
@router.put("/categories/{id_categorie}")
|
||||
def update_categorie(id_categorie: int, body: CategorieBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"UPDATE CATEGORIE SET intitule_categorie = ? WHERE id_categorie = ?",
|
||||
body.intitule_categorie, id_categorie
|
||||
)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Catégorie {id_categorie} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.UPDATE_REFERENTIEL,
|
||||
f"CATEGORIE {id_categorie} : {body.intitule_categorie}")
|
||||
return {"id_categorie": id_categorie, "intitule_categorie": body.intitule_categorie}
|
||||
|
||||
|
||||
@router.delete("/categories/{id_categorie}")
|
||||
def delete_categorie(id_categorie: int, admin: dict = Depends(require_admin)):
|
||||
"""
|
||||
Supprime une catégorie, sauf si des monitorings s'y rattachent encore :
|
||||
la clé étrangère FK_NOM_CATEGORIE l'interdirait, autant renvoyer un
|
||||
message explicite plutôt que de laisser remonter l'erreur SQL brute.
|
||||
"""
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT COUNT(*) FROM NOMENCLATURE_MONITO WHERE id_categorie = ?", id_categorie
|
||||
)
|
||||
nb_rattaches = cursor.fetchone()[0]
|
||||
if nb_rattaches:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=f"Suppression impossible : {nb_rattaches} monitoring(s) utilisent cette catégorie."
|
||||
)
|
||||
cursor.execute("DELETE FROM CATEGORIE WHERE id_categorie = ?", id_categorie)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Catégorie {id_categorie} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.DELETE_REFERENTIEL, f"CATEGORIE {id_categorie}")
|
||||
return {"status": "deleted", "id_categorie": id_categorie}
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Services
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@router.get("/services")
|
||||
def get_services():
|
||||
"""Tous les services."""
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute("SELECT id_service, nom_service FROM SERVICE ORDER BY nom_service")
|
||||
return rows_to_list(cursor, cursor.fetchall())
|
||||
|
||||
|
||||
@router.post("/services", status_code=status.HTTP_201_CREATED)
|
||||
def create_service(body: ServiceBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"INSERT INTO SERVICE (nom_service) OUTPUT INSERTED.id_service VALUES (?)",
|
||||
body.nom_service
|
||||
)
|
||||
id_service = cursor.fetchone()[0]
|
||||
journaliser(cursor, admin, AuditAction.CREATE_REFERENTIEL,
|
||||
f"SERVICE {id_service} : {body.nom_service}")
|
||||
return {"id_service": id_service, "nom_service": body.nom_service}
|
||||
|
||||
|
||||
@router.put("/services/{id_service}")
|
||||
def update_service(id_service: int, body: ServiceBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"UPDATE SERVICE SET nom_service = ? WHERE id_service = ?",
|
||||
body.nom_service, id_service
|
||||
)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Service {id_service} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.UPDATE_REFERENTIEL,
|
||||
f"SERVICE {id_service} : {body.nom_service}")
|
||||
return {"id_service": id_service, "nom_service": body.nom_service}
|
||||
|
||||
|
||||
@router.delete("/services/{id_service}")
|
||||
def delete_service(id_service: int, admin: dict = Depends(require_admin)):
|
||||
"""Refuse la suppression tant que des monitorings ou contacts y sont rattachés."""
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT "
|
||||
"(SELECT COUNT(*) FROM NOMENCLATURE_MONITO WHERE id_service = ?), "
|
||||
"(SELECT COUNT(*) FROM CONTACT WHERE id_service = ?)",
|
||||
id_service, id_service
|
||||
)
|
||||
nb_monitorings, nb_contacts = cursor.fetchone()
|
||||
if nb_monitorings or nb_contacts:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=(f"Suppression impossible : {nb_monitorings} monitoring(s) "
|
||||
f"et {nb_contacts} contact(s) rattachés à ce service.")
|
||||
)
|
||||
cursor.execute("DELETE FROM SERVICE WHERE id_service = ?", id_service)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Service {id_service} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.DELETE_REFERENTIEL, f"SERVICE {id_service}")
|
||||
return {"status": "deleted", "id_service": id_service}
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# Contacts
|
||||
# ------------------------------------------------------------
|
||||
|
||||
@router.get("/contacts")
|
||||
def get_contacts(
|
||||
id_service: Optional[int] = Query(None, description="Filtrer par service")
|
||||
):
|
||||
"""Contacts, filtrables par service."""
|
||||
with get_cursor() as cursor:
|
||||
if id_service:
|
||||
cursor.execute(
|
||||
"SELECT id_contact, id_service, intitule_contact, nom, prenom, mail "
|
||||
"FROM CONTACT WHERE id_service = ? ORDER BY nom",
|
||||
id_service
|
||||
)
|
||||
else:
|
||||
cursor.execute(
|
||||
"SELECT id_contact, id_service, intitule_contact, nom, prenom, mail "
|
||||
"FROM CONTACT ORDER BY nom"
|
||||
)
|
||||
return rows_to_list(cursor, cursor.fetchall())
|
||||
|
||||
|
||||
@router.post("/contacts", status_code=status.HTTP_201_CREATED)
|
||||
def create_contact(body: ContactBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"INSERT INTO CONTACT (id_service, intitule_contact, nom, prenom, mail) "
|
||||
"OUTPUT INSERTED.id_contact VALUES (?, ?, ?, ?, ?)",
|
||||
body.id_service, body.intitule_contact, body.nom, body.prenom, body.mail
|
||||
)
|
||||
id_contact = cursor.fetchone()[0]
|
||||
journaliser(cursor, admin, AuditAction.CREATE_REFERENTIEL,
|
||||
f"CONTACT {id_contact} : {body.nom} {body.prenom}")
|
||||
return {"id_contact": id_contact, **body.model_dump()}
|
||||
|
||||
|
||||
@router.put("/contacts/{id_contact}")
|
||||
def update_contact(id_contact: int, body: ContactBody, admin: dict = Depends(require_admin)):
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute(
|
||||
"UPDATE CONTACT SET id_service = ?, intitule_contact = ?, nom = ?, prenom = ?, mail = ? "
|
||||
"WHERE id_contact = ?",
|
||||
body.id_service, body.intitule_contact, body.nom, body.prenom, body.mail, id_contact
|
||||
)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Contact {id_contact} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.UPDATE_REFERENTIEL,
|
||||
f"CONTACT {id_contact} : {body.nom} {body.prenom}")
|
||||
return {"id_contact": id_contact, **body.model_dump()}
|
||||
|
||||
|
||||
@router.delete("/contacts/{id_contact}")
|
||||
def delete_contact(id_contact: int, admin: dict = Depends(require_admin)):
|
||||
"""Un contact n'est référencé par aucune autre table : suppression directe."""
|
||||
with get_cursor() as cursor:
|
||||
cursor.execute("DELETE FROM CONTACT WHERE id_contact = ?", id_contact)
|
||||
if cursor.rowcount == 0:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Contact {id_contact} introuvable.")
|
||||
journaliser(cursor, admin, AuditAction.DELETE_REFERENTIEL, f"CONTACT {id_contact}")
|
||||
return {"status": "deleted", "id_contact": id_contact}
|
||||
Reference in New Issue
Block a user