groupe-meditation/backend/app/core/roles.py

"""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