carsten
ded23efc6a
Full CRUD on Participant scoped to the caller's organization via new get_route_in_organization/get_participant_in_organization helpers in app/dependencies.py (shared cross-cutting spot, reused by task 6). PATCH is_active is the primary deactivation path — replaces the old spreadsheet-era "empty order" trick and preserves order history for participants who leave and later return. DELETE is a separate, superadmin-gated hard-delete for GDPR erasure requests, which cascades to the participant's orders. Adds UNIQUE(route_id, email) at the DB level (existing 308 participants already conform, per task 3's dedup) plus email normalization on create/update, so case-variant duplicates can't reappear via the API. Every write path relies on catching the constraint's IntegrityError for a clean 409 rather than a racy pre-check. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlmodel import Session, select
|
|
|
|
from app.core.security import decode_access_token
|
|
from app.db import SessionDep
|
|
from app.models.participant import Participant
|
|
from app.models.route import Route
|
|
from app.models.user import User
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
|
|
|
|
|
|
def get_current_user(session: SessionDep, token: Annotated[str, Depends(oauth2_scheme)]) -> User:
|
|
credentials_exception = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
payload = decode_access_token(token)
|
|
if payload is None or payload.get("sub") is None:
|
|
raise credentials_exception
|
|
user = session.get(User, int(payload["sub"]))
|
|
if user is None or not user.is_active:
|
|
raise credentials_exception
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_user)]
|
|
|
|
|
|
def get_current_active_superuser(current_user: CurrentUser) -> User:
|
|
if not current_user.is_superadmin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough privileges")
|
|
return current_user
|
|
|
|
|
|
CurrentSuperuser = Annotated[User, Depends(get_current_active_superuser)]
|
|
|
|
|
|
def get_route_in_organization(session: Session, route_id: int, organization_id: int) -> Route:
|
|
route = session.get(Route, route_id)
|
|
if route is None or route.organization_id != organization_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Route not found")
|
|
return route
|
|
|
|
|
|
def get_participant_in_organization(session: Session, participant_id: int, organization_id: int) -> Participant:
|
|
participant = session.exec(
|
|
select(Participant)
|
|
.join(Route)
|
|
.where(Participant.id == participant_id, Route.organization_id == organization_id)
|
|
).first()
|
|
if participant is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Participant not found")
|
|
return participant
|