feat(identity): IDENT-009 QR-/Code-Neuvergabe mit Historie
CI / backend-tests (push) Successful in 2m20s
CI / frontend-build (push) Successful in 27s

POST /objekte/{id}/code-neu-vergeben und /objektpositionen/{id}/code-neu-
vergeben (admin-only) - z.B. bei beschädigtem/verlorenem Etikett. Alter
Code bleibt über die bestehende Audit-Infrastruktur (Historie-Tabelle,
Prompt 13) nachvollziehbar statt eine eigene Versions-Tabelle zu brauchen:
GET /historie?entitaet_typ=objekt&entitaet_id=... zeigt alter_wert/
neuer_wert des Codes. Duplikat-Prüfung explizit vorab (Projekt-Konvention,
siehe geraet_instanz.py::erstelle).

Frontend: "Code neu vergeben"-Aktion in der Objektakte (nur Admin), inline
statt window.prompt (Accessibility-Vorgabe aus dieser Session).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-08 18:13:53 +02:00
co-authored by Claude Sonnet 5
parent 08ce8efe55
commit b7d9581b90
6 changed files with 262 additions and 2 deletions
+91
View File
@@ -19,6 +19,7 @@ from app.services.label import generiere_code128_label_pdf, generiere_qr_label_p
LabelFormat = Literal["code128", "qr"]
from app.schemas.objekt import (
CodeNeuVergeben,
FahrzeugdetailsRead,
FahrzeugdetailsUpsert,
ObjektCreate,
@@ -209,6 +210,96 @@ async def objektposition_label(
)
@router.post("/objekte/{objekt_id}/code-neu-vergeben", response_model=ObjektRead)
async def objekt_code_neu_vergeben(
objekt_id: int,
payload: CodeNeuVergeben,
db: AsyncSession = Depends(get_db),
current_user=Depends(_admin_only),
) -> ObjektRead:
"""IDENT-009: QR-/Barcode-Neuvergabe, z.B. bei beschädigtem oder verlorenem
Etikett - der alte Code bleibt über die Historie nachvollziehbar (GET
/historie?entitaet_typ=objekt&entitaet_id=...), keine eigene Versions-
Tabelle nötig (bestehende Audit-Infrastruktur, Prompt 13)."""
objekt = await db.get(Objekt, objekt_id)
if objekt is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
bestehendes = await db.execute(select(Objekt).where(Objekt.code == payload.neuer_code, Objekt.id != objekt_id))
if bestehendes.scalar_one_or_none() is not None:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Code bereits vergeben")
alter_code = objekt.code
objekt.code = payload.neuer_code
await db.flush()
hauptserver = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
await historie_service.log(
db,
zustaendiger_server_id=hauptserver.scalar_one(),
benutzer_id=current_user.id,
ereignistyp="objekt_code_neu_vergeben",
entitaet_typ="objekt",
entitaet_id=objekt.id,
alter_wert={"code": alter_code},
neuer_wert={"code": payload.neuer_code},
begruendung=payload.begruendung,
)
letzte_map = await letzte_kontrollen_map(db, [objekt.id])
return _objekt_read(objekt, letzte_map.get(objekt.id))
@router.post("/objektpositionen/{position_id}/code-neu-vergeben", response_model=ObjektpositionRead)
async def objektposition_code_neu_vergeben(
position_id: uuid.UUID,
payload: CodeNeuVergeben,
db: AsyncSession = Depends(get_db),
current_user=Depends(_admin_only),
) -> ObjektpositionRead:
"""IDENT-009: wie objekt_code_neu_vergeben, für Geräte-Etiketten auf
Positionsebene."""
position = await db.get(Objektposition, position_id)
if position is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objektposition nicht gefunden")
bestehendes = await db.execute(
select(Objektposition).where(
Objektposition.code == payload.neuer_code, Objektposition.id != position_id
)
)
if bestehendes.scalar_one_or_none() is not None:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Code bereits vergeben")
alter_code = position.code
position.code = payload.neuer_code
await db.flush()
hauptserver = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
await historie_service.log(
db,
zustaendiger_server_id=hauptserver.scalar_one(),
benutzer_id=current_user.id,
ereignistyp="objektposition_code_neu_vergeben",
entitaet_typ="objektposition",
entitaet_id=position.id,
alter_wert={"code": alter_code},
neuer_wert={"code": payload.neuer_code},
begruendung=payload.begruendung,
)
return ObjektpositionRead(
id=position.id,
objekt_id=position.objekt_id,
material_id=position.material_id,
sollmenge_override=position.sollmenge_override,
sollmenge_effektiv=sollmenge_effektiv(position),
ist_status=position.ist_status,
istmenge=position.istmenge,
seriennummer=position.seriennummer,
ablaufdatum=position.ablaufdatum,
chargennummer=position.chargennummer,
code=position.code,
pruefintervall_monate=position.pruefintervall_monate,
)
@router.get("/objekte/{objekt_id}/positionen", response_model=list[ObjektpositionRead])
async def liste_objektpositionen(
objekt_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)