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.
105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
# ============================================================
|
|
# routers/admin.py — Gestion des comptes et journal d'audit
|
|
#
|
|
# Réservé au rôle Admin (403 sinon), via require_admin.
|
|
# ============================================================
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel
|
|
|
|
from auth import hash_password, require_admin
|
|
from config import get_cursor
|
|
from domain import AuditAction, UserRole
|
|
from helpers import journaliser, rows_to_list
|
|
|
|
router = APIRouter(tags=["Admin"])
|
|
|
|
|
|
class UserCreate(BaseModel):
|
|
username: str
|
|
email: str
|
|
password: str
|
|
role: UserRole
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
email: Optional[str] = None
|
|
role: Optional[UserRole] = None
|
|
actif: Optional[bool] = None
|
|
|
|
|
|
class PasswordReset(BaseModel):
|
|
password: str
|
|
|
|
|
|
@router.get("/admin/users")
|
|
def list_users(admin: dict = Depends(require_admin)):
|
|
with get_cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id_user, username, email, role, actif, created_at, last_login "
|
|
"FROM [USER] ORDER BY id_user"
|
|
)
|
|
return rows_to_list(cursor, cursor.fetchall())
|
|
|
|
|
|
@router.post("/admin/users", status_code=status.HTTP_201_CREATED)
|
|
def create_user(body: UserCreate, admin: dict = Depends(require_admin)):
|
|
# Le rôle est validé par Pydantic via l'enum UserRole (422 si invalide).
|
|
with get_cursor() as cursor:
|
|
cursor.execute(
|
|
"INSERT INTO [USER] (username, email, password_hash, role) VALUES (?, ?, ?, ?)",
|
|
body.username, body.email, hash_password(body.password), body.role.value
|
|
)
|
|
journaliser(cursor, admin, AuditAction.CREATE_USER, body.username)
|
|
return {"status": "created", "username": body.username}
|
|
|
|
|
|
@router.put("/admin/users/{id_user}")
|
|
def update_user(id_user: int, body: UserUpdate, admin: dict = Depends(require_admin)):
|
|
sets, params = [], []
|
|
if body.email is not None:
|
|
sets.append("email = ?"); params.append(body.email)
|
|
if body.role is not None:
|
|
sets.append("role = ?"); params.append(body.role.value)
|
|
if body.actif is not None:
|
|
sets.append("actif = ?"); params.append(1 if body.actif else 0)
|
|
if not sets:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="Aucun champ à modifier"
|
|
)
|
|
params.append(id_user)
|
|
with get_cursor() as cursor:
|
|
cursor.execute(f"UPDATE [USER] SET {', '.join(sets)} WHERE id_user = ?", *params)
|
|
journaliser(cursor, admin, AuditAction.UPDATE_USER, str(id_user))
|
|
return {"status": "updated", "id_user": id_user}
|
|
|
|
|
|
@router.delete("/admin/users/{id_user}")
|
|
def delete_user(id_user: int, admin: dict = Depends(require_admin)):
|
|
"""Suppression douce (actif = 0)."""
|
|
with get_cursor() as cursor:
|
|
cursor.execute("UPDATE [USER] SET actif = 0 WHERE id_user = ?", id_user)
|
|
journaliser(cursor, admin, AuditAction.DELETE_USER, str(id_user))
|
|
return {"status": "deactivated", "id_user": id_user}
|
|
|
|
|
|
@router.post("/admin/users/{id_user}/reset-password")
|
|
def reset_password(id_user: int, body: PasswordReset, admin: dict = Depends(require_admin)):
|
|
with get_cursor() as cursor:
|
|
cursor.execute("UPDATE [USER] SET password_hash = ? WHERE id_user = ?",
|
|
hash_password(body.password), id_user)
|
|
journaliser(cursor, admin, AuditAction.RESET_PASSWORD, str(id_user))
|
|
return {"status": "password_reset", "id_user": id_user}
|
|
|
|
|
|
@router.get("/admin/journal")
|
|
def get_journal(limit: int = 200, admin: dict = Depends(require_admin)):
|
|
with get_cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT TOP (?) date_action, username, action, detail, ip "
|
|
"FROM JOURNAL_AUDIT ORDER BY date_action DESC", limit
|
|
)
|
|
return rows_to_list(cursor, cursor.fetchall())
|