patrickandClaude Sonnet 5
c56fc285de
Additiv neben dem bestehenden 4-Rollen-System (mitarbeiter/material-
verantwortlicher/leitungsverantwortlicher/administration bleiben unverändert,
kein Breaking Change): Admin kann jetzt eigene Rollen (z.B. "Materialwart",
"Helfer" aus der Ursprungs-Anforderung) mit frei wählbaren Einzelrechten aus
einem Berechtigungs-Katalog anlegen und Benutzern zuweisen - auch Benutzern
ganz ohne feste RolleTyp-Zuordnung.
Neue Tabellen: berechtigung (Katalog), rolle (custom, admin-anlegbar),
rolle_berechtigung (M:N), benutzer_rolle_zuordnung (M:N, eigene Tabelle statt
Wiederverwendung des ENUM-basierten benutzer_rolle).
require_roles_or_permission() kombiniert beide Systeme: bestehende feste
Rollen ODER eine passende granulare Berechtigung. Auf die vom Nutzer genannten
Beispiel-Endpunkte angewendet: Material anlegen/bearbeiten, Lagerbewegung
durchführen, Mangel melden/lesen/bearbeiten, Prüfung durchführen - weitere
Endpunkte folgen bei Bedarf nach demselben Muster (require_permission()/
require_roles_or_permission() stehen jetzt als Bausteine bereit).
Neue Endpunkte: GET /berechtigungen, CRUD /rollen, PUT/DELETE
/rollen/{id}/berechtigungen/{id}, PUT/DELETE /benutzer/{id}/rollen/{id}.
Frontend: neuer Admin-Tab "Rollen & Rechte" (Rolle anlegen, Rechte togglen,
Benutzer zuweisen/entfernen).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KC8HYvv6UkCVYheYiTw9DD
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import require_roles, require_roles_or_permission
|
|
from app.db.session import get_db
|
|
from app.models.auth import KnotenTyp, RolleTyp, Systemknoten
|
|
from app.models.geraet_instanz import GeraetInstanz
|
|
from app.models.objektposition import Objektposition
|
|
from app.schemas.geraet_instanz import GeraetInstanzCreate, GeraetInstanzRead, GeraetInstanzStatusUpdate
|
|
from app.services.geraet_instanz import aendere_status, erstelle, liste_fuer_position, loesche
|
|
|
|
router = APIRouter()
|
|
|
|
_materialverantwortliche = require_roles(
|
|
RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
|
|
)
|
|
# Roadmap Phase 6: Prüfung durchführen zusätzlich über die granulare
|
|
# Berechtigung "pruefung.durchfuehren" erreichbar (z.B. custom "Helfer"-Rolle).
|
|
_darf_pruefen = require_roles_or_permission(
|
|
RolleTyp.administration,
|
|
RolleTyp.materialverantwortlicher,
|
|
RolleTyp.leitungsverantwortlicher,
|
|
berechtigung="pruefung.durchfuehren",
|
|
)
|
|
|
|
|
|
async def _hauptserver_id(db: AsyncSession) -> int:
|
|
result = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
|
|
return result.scalar_one()
|
|
|
|
|
|
@router.get("/objektpositionen/{position_id}/geraete", response_model=list[GeraetInstanzRead])
|
|
async def liste_geraete(
|
|
position_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
|
|
) -> list[GeraetInstanz]:
|
|
return await liste_fuer_position(db, position_id)
|
|
|
|
|
|
@router.post(
|
|
"/objektpositionen/{position_id}/geraete",
|
|
response_model=GeraetInstanzRead,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def erstelle_geraet(
|
|
position_id: uuid.UUID,
|
|
payload: GeraetInstanzCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_=Depends(_materialverantwortliche),
|
|
) -> GeraetInstanz:
|
|
"""Karte 14: löst objektposition.seriennummer als Einzelfeld ab - beliebig
|
|
viele Geräte-Exemplare pro Position."""
|
|
position = await db.get(Objektposition, position_id)
|
|
if position is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objektposition nicht gefunden")
|
|
try:
|
|
return await erstelle(db, objektposition_id=position_id, seriennummer=payload.seriennummer)
|
|
except IntegrityError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT, detail="Seriennummer an dieser Position bereits vergeben"
|
|
) from exc
|
|
|
|
|
|
@router.patch("/geraete/{geraet_id}", response_model=GeraetInstanzRead)
|
|
async def aendere_geraet_status(
|
|
geraet_id: uuid.UUID,
|
|
payload: GeraetInstanzStatusUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user=Depends(_darf_pruefen),
|
|
) -> GeraetInstanz:
|
|
instanz = await db.get(GeraetInstanz, geraet_id)
|
|
if instanz is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
|
|
zustaendiger_server_id = await _hauptserver_id(db)
|
|
aktualisiert, _fehlbestand_id = await aendere_status(
|
|
db,
|
|
instanz=instanz,
|
|
status=payload.status,
|
|
pruefdatum=payload.pruefdatum,
|
|
bemerkung=payload.bemerkung,
|
|
erzeuge_fehlbestand=payload.erzeuge_fehlbestand,
|
|
benutzer_id=current_user.id,
|
|
zustaendiger_server_id=zustaendiger_server_id,
|
|
)
|
|
return aktualisiert
|
|
|
|
|
|
@router.delete("/geraete/{geraet_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def loesche_geraet(
|
|
geraet_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
|
|
) -> None:
|
|
instanz = await db.get(GeraetInstanz, geraet_id)
|
|
if instanz is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
await loesche(db, instanz)
|