Daniel Allaire
393f9243cb
- scoper les postes et permissions par groupe - retirer les lectures district du calendrier, des evenements et anniversaires - documenter les regles d isolation multi-groupe
85 lines
3 KiB
Python
85 lines
3 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
|
|
from app.models.poste_module import PosteModule
|
|
|
|
|
|
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
|
|
if not membre:
|
|
return False
|
|
result = await db.execute(
|
|
select(Affectation).join(Poste).where(
|
|
Affectation.membre_id == membre_id,
|
|
Affectation.groupe_id == membre.groupe_id,
|
|
Affectation.statut == "actif",
|
|
Poste.groupe_id == membre.groupe_id,
|
|
(Poste.nom.ilike(f"%{libelle_poste}%") | (Poste.categorie == "executif")),
|
|
)
|
|
)
|
|
return result.scalars().first() is not None
|
|
|
|
|
|
async def membre_a_module(membre_id, module_code: 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
|
|
if not membre:
|
|
return False
|
|
result = await db.execute(
|
|
select(Affectation).join(Poste).join(PosteModule, PosteModule.poste_id == Poste.id).where(
|
|
Affectation.membre_id == membre_id,
|
|
Affectation.groupe_id == membre.groupe_id,
|
|
Affectation.statut == "actif",
|
|
Poste.groupe_id == membre.groupe_id,
|
|
PosteModule.module_code == module_code,
|
|
)
|
|
)
|
|
return result.scalars().first() is not None
|
|
|
|
|
|
async def membre_a_poste_ou_module(membre_id, libelle_poste: str, module_code: str, db: AsyncSession) -> bool:
|
|
return (
|
|
await membre_a_poste_ou_executif(membre_id, libelle_poste, db)
|
|
or await membre_a_module(membre_id, module_code, db)
|
|
)
|
|
|
|
|
|
async def require_tresorier(
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not await membre_a_poste_ou_module(membre.id, "trésorier", "S1T", 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_module(membre.id, "secrétaire", "M1", db)
|
|
or await membre_a_module(membre.id, "M2", 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_module(membre.id, "RSG", "M3", db):
|
|
raise HTTPException(status_code=403, detail="Réservé au RSG ou à l'exécutif")
|
|
return membre
|