95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""Service d'historique métier."""
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import desc, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.concurrency import version_iso
|
|
from app.models.historique_modification import HistoriqueModification
|
|
from app.models.membre import Membre
|
|
|
|
|
|
def snapshot(obj, *champs: str) -> dict:
|
|
"""Extrait un état JSON simple d'un objet métier."""
|
|
data = {}
|
|
for champ in champs:
|
|
valeur = getattr(obj, champ, None)
|
|
if hasattr(valeur, "isoformat"):
|
|
valeur = valeur.isoformat()
|
|
else:
|
|
valeur = str(valeur) if valeur is not None and not isinstance(valeur, (str, int, float, bool, list, dict)) else valeur
|
|
data[champ] = valeur
|
|
return data
|
|
|
|
|
|
async def enregistrer_historique(
|
|
db: AsyncSession,
|
|
*,
|
|
groupe_id,
|
|
table_cible: str,
|
|
objet_id,
|
|
action: str,
|
|
raison: str,
|
|
membre_id=None,
|
|
admin_id=None,
|
|
chemin_api: str | None = None,
|
|
avant: dict | None = None,
|
|
apres: dict | None = None,
|
|
) -> HistoriqueModification:
|
|
entree = HistoriqueModification(
|
|
groupe_id=groupe_id,
|
|
table_cible=table_cible,
|
|
objet_id=objet_id,
|
|
action=action,
|
|
raison=raison,
|
|
membre_id=membre_id,
|
|
admin_id=admin_id,
|
|
chemin_api=chemin_api,
|
|
avant=avant,
|
|
apres=apres,
|
|
)
|
|
db.add(entree)
|
|
return entree
|
|
|
|
|
|
async def dernier_historique(db: AsyncSession, table_cible: str, objet_id) -> HistoriqueModification | None:
|
|
result = await db.execute(
|
|
select(HistoriqueModification)
|
|
.where(
|
|
HistoriqueModification.table_cible == table_cible,
|
|
HistoriqueModification.objet_id == objet_id,
|
|
)
|
|
.order_by(desc(HistoriqueModification.cree_le))
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def verifier_version_historisee(
|
|
db: AsyncSession,
|
|
obj,
|
|
version_attendue: str | None,
|
|
*,
|
|
table_cible: str,
|
|
objet_id=None,
|
|
):
|
|
if not version_attendue:
|
|
return
|
|
actuelle = version_iso(obj)
|
|
if not actuelle or actuelle.replace("+00:00", "Z") == version_attendue.replace("+00:00", "Z"):
|
|
return
|
|
|
|
objet = objet_id or getattr(obj, "id", None)
|
|
entree = await dernier_historique(db, table_cible, objet) if objet else None
|
|
if not entree:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="Cette information a été modifiée depuis son affichage. Rechargez les données avant d'enregistrer.",
|
|
)
|
|
|
|
membre = await db.get(Membre, entree.membre_id) if entree.membre_id else None
|
|
par = f" par {membre.prenom}" if membre else ""
|
|
quand = entree.cree_le.astimezone().strftime("%Y-%m-%d %H:%M")
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Cette information a été modifiée{par} le {quand} : {entree.raison}. Rechargez les données avant d'enregistrer.",
|
|
)
|