feat(akte): Digital-File-Epic - Aggregations-Endpunkt + Übersichtsseite (FILE-001..006)
CI / backend-tests (push) Successful in 1m58s
CI / frontend-build (push) Successful in 18s

Größte Lücke aus arbeitskacheln/03_digital_file.md: alle Teilbereiche
(Stammdaten, Standort, Verantwortlichkeit, Historie, Dokumente, Geräte-
Prüfungen, Mängel) existierten einzeln, aber keine gebündelte Akte-Ansicht
je Objekt.

- GET /api/v1/akte/objekt/{id}: aggregiert alle Teilbereiche in einem Aufruf
  (app/services/akte.py), 404 bei unbekanntem Objekt
- Zuständigkeits-Auflösung invertiert (wer ist für DIESES Objekt zuständig,
  Standort-Vererbung + Objekt-Zeile als Vereinigung, Karte 04)
- Frontend: /akte/objekt/:objektId (AktePage.tsx), Link aus der Objektliste
- 2 neue Tests (test_akte.py), 137 Tests weiterhin grün

FILE-007 (Audit-Konsistenz-Check über alle Mutations-Endpunkte) bewusst
nicht Teil dieses Commits - eigene, spätere Qualitätssicherungs-Kachel.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-05 21:13:29 +02:00
co-authored by Claude Sonnet 5
parent fca9ee8fd8
commit 84c11af3bb
9 changed files with 444 additions and 0 deletions
+134
View File
@@ -0,0 +1,134 @@
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.auth import Benutzer
from app.models.dokument import Dokument
from app.models.fahrzeugdetails import Fahrzeugdetails
from app.models.geraet_instanz import GeraetInstanz
from app.models.historie import Historie
from app.models.mangel import Mangel
from app.models.objekt import Objekt
from app.models.objektposition import Objektposition
from app.models.stammdaten import Objekttyp, Standort
from app.models.zustaendigkeit import Kontrollverantwortung, Zustaendigkeit
from app.schemas.akte import AkteRead, ZustaendigerBenutzer
from app.schemas.dokument import DokumentRead
from app.schemas.geraet_instanz import GeraetInstanzRead
from app.schemas.historie import HistorieRead
from app.schemas.mangel import MangelRead
from app.schemas.objekt import FahrzeugdetailsRead, ObjektRead
from app.schemas.zustaendigkeit import KontrollverantwortungRead
from app.services.objekte import letzte_kontrollen_map
async def _zustaendige_benutzer(db: AsyncSession, objekt: Objekt) -> list[ZustaendigerBenutzer]:
"""Karte 04: Umkehrung von zustaendige_objekte() - wer ist für DIESES Objekt
zuständig (Vereinigung Standort-Vererbung + objektspezifische Zeile)."""
stmt = (
select(Benutzer.id, Benutzer.name, Zustaendigkeit.standort_id, Zustaendigkeit.objekt_id)
.join(
Zustaendigkeit,
or_(
Zustaendigkeit.standort_id == objekt.standort_id,
Zustaendigkeit.objekt_id == objekt.id,
),
)
.where(Benutzer.id == Zustaendigkeit.benutzer_id)
)
result = await db.execute(stmt)
gesehen: set[int] = set()
ergebnis: list[ZustaendigerBenutzer] = []
for benutzer_id, name, standort_id, objekt_id in result.all():
if benutzer_id in gesehen:
continue
gesehen.add(benutzer_id)
quelle = "objekt" if objekt_id == objekt.id else "standort"
ergebnis.append(ZustaendigerBenutzer(id=benutzer_id, name=name, quelle=quelle))
return ergebnis
async def akte_fuer_objekt(db: AsyncSession, objekt_id: int) -> AkteRead | None:
"""FILE-001: bündelt Stammdaten/Standort/Verantwortlichkeit/Historie/
Dokumente/Prüfungen/Mängel für ein Objekt zu einer Aggregations-Sicht."""
objekt = await db.get(Objekt, objekt_id)
if objekt is None:
return None
objekttyp = await db.get(Objekttyp, objekt.objekttyp_id)
standort = await db.get(Standort, objekt.standort_id)
fahrzeugdetails = await db.get(Fahrzeugdetails, objekt_id)
letzte_map = await letzte_kontrollen_map(db, [objekt_id])
zustaendige = await _zustaendige_benutzer(db, objekt)
kontrollverantwortung_result = await db.execute(
select(Kontrollverantwortung).where(Kontrollverantwortung.objekt_id == objekt_id)
)
historie_result = await db.execute(
select(Historie, Benutzer.name)
.outerjoin(Benutzer, Benutzer.id == Historie.benutzer_id)
.where(Historie.entitaet_typ == "objekt", Historie.entitaet_id == str(objekt_id))
.order_by(Historie.zeitpunkt.desc())
.limit(50)
)
dokumente_result = await db.execute(
select(Dokument).where(Dokument.entitaet_typ == "objekt", Dokument.entitaet_id == str(objekt_id))
)
positions_result = await db.execute(
select(Objektposition.id).where(Objektposition.objekt_id == objekt_id)
)
position_ids = [row[0] for row in positions_result.all()]
geraete: list[GeraetInstanz] = []
if position_ids:
geraete_result = await db.execute(
select(GeraetInstanz).where(GeraetInstanz.objektposition_id.in_(position_ids))
)
geraete = list(geraete_result.scalars().all())
maengel_result = await db.execute(select(Mangel).where(Mangel.objekt_id == objekt_id))
objekt_read = ObjektRead(
id=objekt.id,
code=objekt.code,
name=objekt.name,
objekttyp_id=objekt.objekttyp_id,
vorlage_id=objekt.vorlage_id,
standort_id=objekt.standort_id,
status=objekt.status,
fahrzeug_id=objekt.fahrzeug_id,
letzte_kontrolle_am=letzte_map.get(objekt_id),
)
return AkteRead(
objekt=objekt_read,
objekttyp_name=objekttyp.name if objekttyp else "",
standort_name=standort.name if standort else "",
fahrzeugdetails=FahrzeugdetailsRead.model_validate(fahrzeugdetails) if fahrzeugdetails else None,
zustaendige_benutzer=zustaendige,
kontrollverantwortung=[
KontrollverantwortungRead.model_validate(k) for k in kontrollverantwortung_result.scalars().all()
],
letzte_kontrolle_am=letzte_map.get(objekt_id),
historie=[
HistorieRead(
id=eintrag.id,
zeitpunkt=eintrag.zeitpunkt,
benutzer_id=eintrag.benutzer_id,
benutzer_name=benutzer_name,
ereignistyp=eintrag.ereignistyp,
entitaet_typ=eintrag.entitaet_typ,
entitaet_id=eintrag.entitaet_id,
alter_wert=eintrag.alter_wert,
neuer_wert=eintrag.neuer_wert,
begruendung=eintrag.begruendung,
)
for eintrag, benutzer_name in historie_result.all()
],
dokumente=[DokumentRead.model_validate(d) for d in dokumente_result.scalars().all()],
geraete=[GeraetInstanzRead.model_validate(g) for g in geraete],
maengel=[MangelRead.model_validate(m) for m in maengel_result.scalars().all()],
)