patrickandClaude Sonnet 5
fca9ee8fd8
Root Cause der "CI rot"-Meldung gefunden und behoben:
- CI legte Testdatenbank ohne explizites ENCODING an, erbte SQL_ASCII vom
Runner-Postgres-Template. Umlaute in JSONB (Historie-Einträge) brachen mit
UntranslatableCharacterError - betraf auch die Produktions-DB (separat
gemeldet, nicht Teil dieses Commits). CI legt jetzt explizit UTF8 an.
Drei echte Bugs beim Verifizieren gegen eine isolierte Testdatenbank gefunden:
- objekte.py PATCH /objekte/{id}: Selbstbezug-Check verließ sich auf den
DB-CHECK-Constraint statt vorab zu prüfen - ein Flush-Fehlschlag hinterlässt
die Session im Zustand DEACTIVE, jeder folgende Request in derselben Session
crasht mit PendingRollbackError (401 statt 404 im Test). Jetzt expliziter
Vorab-Check.
- geraet_instanz.py: gleiches Muster bei doppelter Seriennummer - jetzt
expliziter Vorab-Check statt UNIQUE-Constraint-Exception.
- test_lagerbewegung.py: Testbug, las objekt.standort_id NACH dem POST (durch
geteilte Session bereits auf den neuen Wert mutiert) statt vorher.
Nebenbei (Auftrag Priorität 4): "Mindermenge genehmigen"-Button direkt im
Nachfüll-Dialog der Kontroll-Erfassung (nur für materialverantwortlicher/
leitungsverantwortlicher/administration), nutzt den bereits bestehenden
POST /fehlbestaende/{id}/mindermenge Endpunkt.
Priorität 2 (Fahrzeug-Feldnamen-Mismatch) und Priorität 3 (nur Zugfahrzeuge
wählbar) waren bereits in früheren Commits erledigt (38b8ce1, 54fc296) -
Auftragsbeschreibung war auf altem Stand.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
105 lines
3.9 KiB
Python
105 lines
3.9 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, 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 (
|
|
SeriennummerBereitsVergebenError,
|
|
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 SeriennummerBereitsVergebenError 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)
|