carsten
acaa2383e9
Full CRUD on PurchaseRound and WineOffering, org-scoped via two new
join-based dependencies.py helpers (get_round_in_organization,
get_wine_offering_in_organization) mirroring task 5's pattern. New
rounds always start as draft regardless of what the client posts,
avoiding a confusing creation-time constraint error.
POST /purchase-rounds/{id}/copy implements "kopiér fra forrige runde":
duplicates every round field (dates, texts) and every wine offering
(incl. category) into a fresh draft, leaving the source untouched.
PurchaseRound DELETE gets a three-tier policy based on status: draft
deletable by any admin, open never deletable, closed only by an
elevated superadmin. This is runtime-conditional on the loaded row, so
the elevation check was extracted out of get_current_active_superuser
into a standalone require_elevated_superuser(user, token) helper that
both the dependency and this handler call directly. WineOffering
delete has no such tier — an offering with real order lines is already
blocked by the existing RESTRICT FK, caught here as a 409.
Also seeds the 8 wine categories (empty table blocked any offering
creation) via a plain Alembic data migration, idempotent and
unconditional (no secret involved, unlike the task-4 superuser seed).
Verified end-to-end: dates-required 409, open round successfully
patched with dates, wine offerings created, round copied (new draft,
duplicated offerings with new ids/same category), open-round delete
403, closed-round delete 403 then 204 after /auth/elevate, wine
offering delete 204, draft round delete 204 (cascades its remaining
offering). DB left clean, participant count unaffected.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
96 lines
3.6 KiB
Python
96 lines
3.6 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.purchase_round import PurchaseRound
|
|
from app.models.route import Route
|
|
from app.models.user import User
|
|
from app.models.wine_offering import WineOffering
|
|
|
|
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 require_elevated_superuser(current_user: User, token: str) -> None:
|
|
if not current_user.is_superadmin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough privileges")
|
|
payload = decode_access_token(token)
|
|
if not payload or not payload.get("elevated"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="This action requires an elevated session — call POST /auth/elevate first",
|
|
)
|
|
|
|
|
|
def get_current_active_superuser(
|
|
current_user: CurrentUser,
|
|
token: Annotated[str, Depends(oauth2_scheme)],
|
|
) -> User:
|
|
require_elevated_superuser(current_user, token)
|
|
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
|
|
|
|
|
|
def get_round_in_organization(session: Session, round_id: int, organization_id: int) -> PurchaseRound:
|
|
purchase_round = session.exec(
|
|
select(PurchaseRound)
|
|
.join(Route)
|
|
.where(PurchaseRound.id == round_id, Route.organization_id == organization_id)
|
|
).first()
|
|
if purchase_round is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Purchase round not found")
|
|
return purchase_round
|
|
|
|
|
|
def get_wine_offering_in_organization(session: Session, wine_offering_id: int, organization_id: int) -> WineOffering:
|
|
wine_offering = session.exec(
|
|
select(WineOffering)
|
|
.join(PurchaseRound)
|
|
.join(Route)
|
|
.where(WineOffering.id == wine_offering_id, Route.organization_id == organization_id)
|
|
).first()
|
|
if wine_offering is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Wine offering not found")
|
|
return wine_offering
|