"""Zugriffsrechte auf Listen und Fortschreibung des Revisionszählers. Rollen, aufsteigend: viewer darf lesen editor darf Einträge, Artikel, Märkte und Warengruppen ändern owner darf zusätzlich die Liste umbenennen, löschen und Mitgliedschaften verwalten Der Eigentümer steht doppelt fest: in `shopping_list.owner_id` und als Mitglied mit der Rolle "owner". Das ist bewusst redundant - die Spalte verhindert, dass eine Liste durch Löschen des letzten Mitglieds verwaist, die Mitgliedschaft macht Abfragen einheitlich. """ from typing import Annotated from fastapi import Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session from app.deps import DbSession, VerifiedUser from app.models import ListMember, ShoppingList ROLE_RANK = {"viewer": 1, "editor": 2, "owner": 3} def bump_rev(db: Session, list_id: str) -> int: """Erhöht den Revisionszähler der Liste und gibt den neuen Wert zurück. SELECT ... FOR UPDATE sperrt die Zeile bis zum Commit. Ohne die Sperre könnten zwei gleichzeitige Änderungen denselben Wert vergeben - dann verpasst ein synchronisierender Client eine davon. """ lst = db.scalar( select(ShoppingList).where(ShoppingList.id == list_id).with_for_update() ) if lst is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "Liste nicht gefunden") lst.rev += 1 db.flush() return lst.rev def _access(db: Session, list_id: str, user_id: str, minimum: str) -> ShoppingList: lst = db.scalar( select(ShoppingList).where( ShoppingList.id == list_id, ShoppingList.deleted_at.is_(None) ) ) membership = db.get(ListMember, (list_id, user_id)) # Gleiche Antwort für "gibt es nicht" und "darfst du nicht sehen". # Sonst ließe sich über die Statuscodes herausfinden, welche # Listen-IDs existieren. if lst is None or membership is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "Liste nicht gefunden") if ROLE_RANK.get(membership.role, 0) < ROLE_RANK[minimum]: raise HTTPException( status.HTTP_403_FORBIDDEN, f"Für diese Aktion ist mindestens die Rolle '{minimum}' erforderlich.", ) return lst def list_reader(list_id: str, db: DbSession, user: VerifiedUser) -> ShoppingList: return _access(db, list_id, user.id, "viewer") def list_editor(list_id: str, db: DbSession, user: VerifiedUser) -> ShoppingList: return _access(db, list_id, user.id, "editor") def list_owner(list_id: str, db: DbSession, user: VerifiedUser) -> ShoppingList: return _access(db, list_id, user.id, "owner") ReadableList = Annotated[ShoppingList, Depends(list_reader)] EditableList = Annotated[ShoppingList, Depends(list_editor)] OwnedList = Annotated[ShoppingList, Depends(list_owner)]