"""Disponibilités déclarées par les bénévoles. Une disponibilité = « le , je peux de à ». Fenêtre horaire libre (pas liée aux créneaux types) ; plusieurs par jour possibles. `end_time <= start_time` ⇒ la fenêtre se termine le lendemain. Ces fenêtres sont prises en compte par le moteur de validation (règle `OUTSIDE_AVAILABILITY`) au moment d'affecter un bénévole à un créneau. """ import uuid from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import delete from sqlmodel import Session, select from app.api._common import owned_or_404 from app.api.deps import get_edition from app.api.schemas import AvailabilityRead, AvailabilitySet from app.auth.deps import ( current_account_in_edition, require_coordinator, require_staff, ) from app.db import get_session from app.models import Account, Availability, Edition, Role, Volunteer router = APIRouter(prefix="/editions/{edition_id}", tags=["availability"]) def _ensure_can_read(account: Account, volunteer_id: uuid.UUID) -> None: if account.role in (Role.COORDINATOR, Role.LEAD): return if account.volunteer_id == volunteer_id: return raise HTTPException(status_code=403, detail={"code": "error.forbidden"}) def _windows(session: Session, volunteer_id: uuid.UUID) -> list[Availability]: return list( session.exec( select(Availability) .where(Availability.volunteer_id == volunteer_id) .order_by(Availability.date, Availability.start_time) ) ) @router.get( "/volunteers/{volunteer_id}/availability", response_model=list[AvailabilityRead], ) def list_availability( volunteer_id: uuid.UUID, edition: Edition = Depends(get_edition), account: Account = Depends(current_account_in_edition), session: Session = Depends(get_session), ) -> list[Availability]: _ensure_can_read(account, volunteer_id) owned_or_404(session.get(Volunteer, volunteer_id), edition.id) return _windows(session, volunteer_id) @router.put( "/volunteers/{volunteer_id}/availability", response_model=list[AvailabilityRead], ) def replace_availability( volunteer_id: uuid.UUID, payload: AvailabilitySet, edition: Edition = Depends(get_edition), _guard: Account = Depends(require_coordinator), session: Session = Depends(get_session), ) -> list[Availability]: owned_or_404(session.get(Volunteer, volunteer_id), edition.id) for w in payload.windows: if not edition.start_date <= w.date <= edition.end_date: raise HTTPException( status_code=422, detail={ "code": "availability.date_outside_edition", "params": {"date": w.date.isoformat()}, }, ) if w.start_time == w.end_time: raise HTTPException( status_code=422, detail={"code": "availability.empty_window"}, ) session.execute( delete(Availability).where(Availability.volunteer_id == volunteer_id) ) # Dé-doublonne (date, start, end) : la contrainte d'unicité l'imposerait. seen: set[tuple] = set() for w in payload.windows: key = (w.date, w.start_time, w.end_time) if key in seen: continue seen.add(key) session.add( Availability( edition_id=edition.id, volunteer_id=volunteer_id, date=w.date, start_time=w.start_time, end_time=w.end_time, ) ) session.commit() return _windows(session, volunteer_id) @router.get("/availability", response_model=list[AvailabilityRead]) def list_edition_availability( edition: Edition = Depends(get_edition), _guard: Account = Depends(require_staff), session: Session = Depends(get_session), ) -> list[Availability]: """Toutes les disponibilités de l'édition (pour la grille des besoins).""" return list( session.exec( select(Availability) .where(Availability.edition_id == edition.id) .order_by( Availability.volunteer_id, Availability.date, Availability.start_time, ) ) )