51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
"""Vérifications de rôles applicatifs."""
|
|
from fastapi import Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_membre, is_sysadmin
|
|
from app.models.affectation import Affectation
|
|
from app.models.poste import Poste
|
|
|
|
|
|
async def membre_a_poste_ou_executif(membre_id, libelle_poste: str, db: AsyncSession) -> bool:
|
|
from app.models.membre import Membre
|
|
membre = await db.get(Membre, membre_id)
|
|
if membre and is_sysadmin(membre):
|
|
return True
|
|
result = await db.execute(
|
|
select(Affectation).join(Poste).where(
|
|
Affectation.membre_id == membre_id,
|
|
Affectation.statut == "actif",
|
|
(Poste.nom.ilike(f"%{libelle_poste}%") | (Poste.categorie == "executif")),
|
|
)
|
|
)
|
|
return result.scalars().first() is not None
|
|
|
|
|
|
async def require_tresorier(
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not await membre_a_poste_ou_executif(membre.id, "trésorier", db):
|
|
raise HTTPException(status_code=403, detail="Réservé au trésorier ou à l'exécutif")
|
|
return membre
|
|
|
|
|
|
async def require_secretaire(
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not await membre_a_poste_ou_executif(membre.id, "secrétaire", db):
|
|
raise HTTPException(status_code=403, detail="Réservé au secrétaire ou à l'exécutif")
|
|
return membre
|
|
|
|
|
|
async def require_rsg(
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not await membre_a_poste_ou_executif(membre.id, "RSG", db):
|
|
raise HTTPException(status_code=403, detail="Réservé au RSG ou à l'exécutif")
|
|
return membre
|