Daniel Allaire
c4c7dc7b64
- retirer les actions BD du profil - ajouter module Gestion des groupes - neutraliser le titre de la page de selection
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""Routes d'administration reservees au sysadmin."""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import func, select
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_membre, is_sysadmin
|
|
from app.services.admin_database import initialiser_demo, remettre_a_zero
|
|
from app.models.groupe import Groupe
|
|
from app.models.membre import Membre
|
|
from app.models.reunion import Reunion
|
|
from app.models.historique_modification import HistoriqueModification
|
|
|
|
router = APIRouter(prefix="/admin", tags=["Administration"])
|
|
|
|
|
|
async def require_sysadmin(membre=Depends(get_current_membre)):
|
|
if not is_sysadmin(membre):
|
|
raise HTTPException(status_code=403, detail="Réservé au sysadmin")
|
|
return membre
|
|
|
|
|
|
@router.post("/reset")
|
|
async def reset_database(
|
|
_sysadmin=Depends(require_sysadmin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
return await remettre_a_zero(db)
|
|
|
|
|
|
@router.post("/demo")
|
|
async def demo_database(
|
|
_sysadmin=Depends(require_sysadmin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
return await initialiser_demo(db)
|
|
|
|
|
|
@router.get("/groupes")
|
|
async def liste_groupes_admin(
|
|
_sysadmin=Depends(require_sysadmin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
groupes = (await db.execute(select(Groupe).order_by(Groupe.nom))).scalars().all()
|
|
out = []
|
|
for groupe in groupes:
|
|
nb_membres = (await db.execute(
|
|
select(func.count()).select_from(Membre).where(Membre.groupe_id == groupe.id)
|
|
)).scalar_one()
|
|
nb_reunions = (await db.execute(
|
|
select(func.count()).select_from(Reunion).where(Reunion.groupe_id == groupe.id)
|
|
)).scalar_one()
|
|
nb_traces = (await db.execute(
|
|
select(func.count()).select_from(HistoriqueModification).where(HistoriqueModification.groupe_id == groupe.id)
|
|
)).scalar_one()
|
|
out.append({
|
|
"id": str(groupe.id),
|
|
"nom": groupe.nom,
|
|
"district": groupe.district,
|
|
"region": groupe.region,
|
|
"adresse_local": groupe.adresse_local,
|
|
"jour_reunion": groupe.jour_reunion,
|
|
"heure_reunion": groupe.heure_reunion.strftime("%H:%M") if groupe.heure_reunion else None,
|
|
"assemblee_affaires_ordre": groupe.assemblee_affaires_ordre,
|
|
"format_reunion": groupe.format_reunion,
|
|
"cree_le": groupe.cree_le,
|
|
"membres": nb_membres,
|
|
"reunions": nb_reunions,
|
|
"traces": nb_traces,
|
|
})
|
|
return out
|