patrickandClaude Sonnet 5
b9cf195a63
CI / backend-tests (push) Failing after 0s
- ORM-Modelle: Bereich, Kategorie, Standort, Objekttyp, Material (Prompt 06/07),
minimales Objekt-Modell (voll ausgebaut erst Sprint 2), Zustaendigkeit/
Kontrollverantwortung
- Admin-API 7a: GET/POST Stammdaten-Endpunkte, PATCH Material, Benutzerverwaltung
(anlegen/Rollen ändern), Zuständigkeits-/Kontrollverantwortungs-CRUD
- Vererbungslogik E2 als eigener Service (app/services/zustaendigkeit.py):
Standort-Zuordnung vererbt sich auf alle Objekte, Objekt-Zeile ist Vereinigung
statt Ersatz - GET /zustaendigkeiten/benutzer/{id}/objekte exponiert das
- Tests: Stammdaten-CRUD + Rollenrechte, Benutzerverwaltung (inkl. 409 bei
Login-Duplikat), E2-Vererbungslogik (Standort-only, Standort+Objekt-Vereinigung,
End-to-End über den Endpunkt)
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import require_roles
|
|
from app.db.session import get_db
|
|
from app.models.auth import RolleTyp
|
|
from app.models.zustaendigkeit import Kontrollverantwortung, Zustaendigkeit
|
|
from app.schemas.zustaendigkeit import (
|
|
KontrollverantwortungCreate,
|
|
KontrollverantwortungRead,
|
|
ObjektKurz,
|
|
ZustaendigkeitCreate,
|
|
ZustaendigkeitRead,
|
|
)
|
|
from app.services.zustaendigkeit import zustaendige_objekte
|
|
|
|
router = APIRouter()
|
|
|
|
_admin_only = require_roles(RolleTyp.administration)
|
|
_verantwortliche = require_roles(
|
|
RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
|
|
)
|
|
|
|
|
|
@router.get("/zustaendigkeiten", response_model=list[ZustaendigkeitRead])
|
|
async def liste_zustaendigkeiten(
|
|
db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
|
|
) -> list[Zustaendigkeit]:
|
|
result = await db.execute(select(Zustaendigkeit))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.post(
|
|
"/zustaendigkeiten", response_model=ZustaendigkeitRead, status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def erstelle_zustaendigkeit(
|
|
payload: ZustaendigkeitCreate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
|
|
) -> Zustaendigkeit:
|
|
zustaendigkeit = Zustaendigkeit(**payload.model_dump())
|
|
db.add(zustaendigkeit)
|
|
await db.flush()
|
|
return zustaendigkeit
|
|
|
|
|
|
@router.delete("/zustaendigkeiten/{zustaendigkeit_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def loesche_zustaendigkeit(
|
|
zustaendigkeit_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
|
|
) -> None:
|
|
zustaendigkeit = await db.get(Zustaendigkeit, zustaendigkeit_id)
|
|
if zustaendigkeit is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Zuständigkeit nicht gefunden")
|
|
await db.delete(zustaendigkeit)
|
|
|
|
|
|
@router.get("/zustaendigkeiten/benutzer/{benutzer_id}/objekte", response_model=list[ObjektKurz])
|
|
async def objekte_eines_verantwortlichen(
|
|
benutzer_id: int, db: AsyncSession = Depends(get_db), _=Depends(_verantwortliche)
|
|
) -> list[ObjektKurz]:
|
|
"""Vererbungslogik E2: Standort-Zuordnung + Objekt-Zuordnung als Vereinigung."""
|
|
objekte = await zustaendige_objekte(db, benutzer_id)
|
|
return [ObjektKurz.model_validate(o) for o in objekte]
|
|
|
|
|
|
@router.get("/kontrollverantwortung", response_model=list[KontrollverantwortungRead])
|
|
async def liste_kontrollverantwortung(
|
|
db: AsyncSession = Depends(get_db), _=Depends(_verantwortliche)
|
|
) -> list[Kontrollverantwortung]:
|
|
result = await db.execute(select(Kontrollverantwortung))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.post(
|
|
"/kontrollverantwortung",
|
|
response_model=KontrollverantwortungRead,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def erstelle_kontrollverantwortung(
|
|
payload: KontrollverantwortungCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_=Depends(_verantwortliche),
|
|
) -> Kontrollverantwortung:
|
|
"""Karte 01 / E3: reine Sichtbarkeit/Priorisierung, erzwingt keine Pflichtkontrolle."""
|
|
kontrollverantwortung = Kontrollverantwortung(**payload.model_dump())
|
|
db.add(kontrollverantwortung)
|
|
await db.flush()
|
|
return kontrollverantwortung
|