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
+2
View File
@@ -1,6 +1,7 @@
from fastapi import APIRouter
from app.api.v1.endpoints import (
akte,
auth,
benutzer,
dashboard,
@@ -40,3 +41,4 @@ api_router.include_router(mangel.router, tags=["mangel"])
api_router.include_router(lagerbewegung.router, tags=["lagerbewegung"])
api_router.include_router(dokument.router, tags=["dokument"])
api_router.include_router(permission.router, tags=["permission"])
api_router.include_router(akte.router, tags=["akte"])
+23
View File
@@ -0,0 +1,23 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.db.session import get_db
from app.schemas.akte import AkteRead
from app.services.akte import akte_fuer_objekt
router = APIRouter()
@router.get("/akte/objekt/{objekt_id}", response_model=AkteRead)
async def hole_akte_fuer_objekt(
objekt_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
) -> AkteRead:
"""FILE-001/006: Aggregations-Endpunkt für die Akte-Übersichtsseite - bündelt
Stammdaten/Standort/Verantwortlichkeit/Historie/Dokumente/Prüfungen/Mängel
eines Objekts in einem Aufruf, statt mehrere Einzelseiten durchklicken zu
müssen (arbeitskacheln/03_digital_file.md)."""
akte = await akte_fuer_objekt(db, objekt_id)
if akte is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
return akte
+36
View File
@@ -0,0 +1,36 @@
from datetime import datetime
from pydantic import BaseModel, ConfigDict
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
class ZustaendigerBenutzer(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
quelle: str # "standort" oder "objekt" (Vereinigungslogik Karte 04)
class AkteRead(BaseModel):
"""FILE-001..006: gebündelte Aggregations-Sicht über alle bereits existierenden
Teilbereiche einer Ressource (aktuell nur Objekttyp "objekt" - Prompt 24/Backlog-
Roadmap sieht perspektivisch weitere Ressourcentypen vor, siehe arbeitskacheln/
03_digital_file.md)."""
objekt: ObjektRead
objekttyp_name: str
standort_name: str
fahrzeugdetails: FahrzeugdetailsRead | None
zustaendige_benutzer: list[ZustaendigerBenutzer]
kontrollverantwortung: list[KontrollverantwortungRead]
letzte_kontrolle_am: datetime | None
historie: list[HistorieRead]
dokumente: list[DokumentRead]
geraete: list[GeraetInstanzRead]
maengel: list[MangelRead]
+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()],
)
+57
View File
@@ -0,0 +1,57 @@
import pytest
from tests.conftest import auth_header, login
@pytest.mark.asyncio
async def test_akte_buendelt_teilbereiche(client, db_session, objekt_mit_position, admin_user, hauptserver_id):
"""FILE-001: Aggregations-Endpunkt liefert Stammdaten + mind. zwei weitere
Teilbereiche (Historie, Mängel) für ein Objekt in einem Aufruf."""
from app.models.mangel import Mangel, MangelPrioritaet, MangelStatus
from app.models.historie import Historie
from datetime import datetime, timezone
import uuid
objekt, _material = objekt_mit_position
db_session.add(
Mangel(
objekt_id=objekt.id,
beschreibung="Riss im Gurt",
prioritaet=MangelPrioritaet.hoch,
status=MangelStatus.neu,
gemeldet_von=admin_user.id,
gemeldet_am=datetime.now(timezone.utc),
)
)
db_session.add(
Historie(
id=uuid.uuid4(),
erzeugt_von_server_id=hauptserver_id,
zeitpunkt=datetime.now(timezone.utc),
benutzer_id=admin_user.id,
ereignistyp="objekt_geaendert",
entitaet_typ="objekt",
entitaet_id=str(objekt.id),
begruendung=None,
)
)
await db_session.flush()
token = await login(client, "admin1")
antwort = await client.get(f"/api/v1/akte/objekt/{objekt.id}", headers=auth_header(token))
assert antwort.status_code == 200
body = antwort.json()
assert body["objekt"]["id"] == objekt.id
assert body["objekttyp_name"] == "Rucksack-KT"
assert body["standort_name"] == "Wache-KT"
assert len(body["maengel"]) == 1
assert body["maengel"][0]["beschreibung"] == "Riss im Gurt"
assert len(body["historie"]) == 1
assert body["historie"][0]["entitaet_typ"] == "objekt"
@pytest.mark.asyncio
async def test_akte_unbekanntes_objekt_ist_404(client, admin_user):
token = await login(client, "admin1")
antwort = await client.get("/api/v1/akte/objekt/999999", headers=auth_header(token))
assert antwort.status_code == 404
+9
View File
@@ -4,6 +4,7 @@ import { Navigate, Route, Routes } from "react-router-dom";
import { AuthProvider, useAuth } from "./auth/AuthContext";
import { AppShell } from "./components/AppShell";
import { AdminPage } from "./pages/AdminPage";
import { AktePage } from "./pages/AktePage";
import { DashboardPage } from "./pages/DashboardPage";
import { FehlbestandListePage } from "./pages/FehlbestandListePage";
import { KontrollPage } from "./pages/KontrollPage";
@@ -46,6 +47,14 @@ export default function App() {
</GeschuetzteRoute>
}
/>
<Route
path="/akte/objekt/:objektId"
element={
<GeschuetzteRoute>
<AktePage />
</GeschuetzteRoute>
}
/>
<Route
path="/objekte/:objektId/kontrolle"
element={
+20
View File
@@ -207,6 +207,26 @@ export interface EskalationKonfiguration {
leitung_tage: number;
}
export interface ZustaendigerBenutzer {
id: number;
name: string;
quelle: "standort" | "objekt";
}
export interface Akte {
objekt: Objekt;
objekttyp_name: string;
standort_name: string;
fahrzeugdetails: Fahrzeugdetails | null;
zustaendige_benutzer: ZustaendigerBenutzer[];
kontrollverantwortung: Kontrollverantwortung[];
letzte_kontrolle_am: string | null;
historie: HistorieEintrag[];
dokumente: Dokument[];
geraete: GeraetInstanz[];
maengel: Mangel[];
}
export interface HistorieEintrag {
id: string;
zeitpunkt: string;
+155
View File
@@ -0,0 +1,155 @@
import { useEffect, useState } from "react";
import { Link, useParams } from "react-router-dom";
import { apiRequest } from "../api/client";
import type { Akte } from "../api/types";
const MANGEL_PRIORITAET_LABEL: Record<string, string> = {
niedrig: "Niedrig",
normal: "Normal",
hoch: "Hoch",
};
// FILE-006: EIN Screen für alle Teilbereiche einer Ressource statt mehrerer
// Einzelseiten - konsumiert den FILE-001-Aggregations-Endpunkt.
export function AktePage() {
const { objektId } = useParams<{ objektId: string }>();
const [akte, setAkte] = useState<Akte | null>(null);
const [fehler, setFehler] = useState<string | null>(null);
const [laedt, setLaedt] = useState(true);
useEffect(() => {
if (!objektId) return;
setLaedt(true);
apiRequest<Akte>(`/akte/objekt/${objektId}`)
.then(setAkte)
.catch(() => setFehler("Akte konnte nicht geladen werden."))
.finally(() => setLaedt(false));
}, [objektId]);
if (laedt) return <main className="page"><p className="text-muted">Lade</p></main>;
if (fehler) return <main className="page"><div className="alert alert-danger">{fehler}</div></main>;
if (!akte) return null;
const { objekt } = akte;
return (
<main className="page">
<div className="row-between" style={{ alignItems: "flex-start" }}>
<div>
<h1 style={{ marginBottom: "0.15rem" }}>{objekt.name}</h1>
<div className="text-muted">
{objekt.code} · {akte.objekttyp_name}
</div>
</div>
<span className="badge badge-neutral">{objekt.status}</span>
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Stammdaten</h3>
<div className="stack">
<div>Standort: {akte.standort_name}</div>
<div>
Letzte Kontrolle:{" "}
{akte.letzte_kontrolle_am
? new Date(akte.letzte_kontrolle_am).toLocaleString("de-DE")
: "noch nie kontrolliert"}
</div>
{akte.fahrzeugdetails && (
<>
{akte.fahrzeugdetails.kennzeichen && <div>Kennzeichen: {akte.fahrzeugdetails.kennzeichen}</div>}
{akte.fahrzeugdetails.funkrufname && <div>Funkrufname: {akte.fahrzeugdetails.funkrufname}</div>}
{akte.fahrzeugdetails.funkkenner && <div>Funkkenner: {akte.fahrzeugdetails.funkkenner}</div>}
{akte.fahrzeugdetails.kilometerstand !== null && (
<div>Kilometerstand: {akte.fahrzeugdetails.kilometerstand} km</div>
)}
</>
)}
</div>
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Verantwortlichkeit</h3>
{akte.zustaendige_benutzer.length === 0 && akte.kontrollverantwortung.length === 0 && (
<p className="text-muted">Keine Zuordnung hinterlegt.</p>
)}
{akte.zustaendige_benutzer.length > 0 && (
<ul>
{akte.zustaendige_benutzer.map((z) => (
<li key={z.id}>
{z.name} <span className="text-muted">({z.quelle === "standort" ? "über Standort" : "direkt"})</span>
</li>
))}
</ul>
)}
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Prüfungen (Geräte)</h3>
{akte.geraete.length === 0 && <p className="text-muted">Keine Geräte-Instanzen erfasst.</p>}
{akte.geraete.length > 0 && (
<ul>
{akte.geraete.map((g) => (
<li key={g.id}>
SN {g.seriennummer} {" "}
<span className={g.status === "einsatzbereit" ? "text-success" : "text-danger"}>{g.status}</span>
{g.naechste_pruefung && (
<span className="text-muted"> · nächste Prüfung {new Date(g.naechste_pruefung).toLocaleDateString("de-DE")}</span>
)}
</li>
))}
</ul>
)}
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Mängel</h3>
{akte.maengel.length === 0 && <p className="text-muted">Keine Mängel gemeldet.</p>}
{akte.maengel.length > 0 && (
<ul>
{akte.maengel.map((m) => (
<li key={m.id}>
{m.beschreibung} {MANGEL_PRIORITAET_LABEL[m.prioritaet]}, {m.status}
</li>
))}
</ul>
)}
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Dokumente</h3>
{akte.dokumente.length === 0 && <p className="text-muted">Keine Dokumente hinterlegt.</p>}
{akte.dokumente.length > 0 && (
<ul>
{akte.dokumente.map((d) => (
<li key={d.id}>
{d.dateiname} <span className="text-muted">({Math.round(d.groesse_bytes / 1024)} KB)</span>
</li>
))}
</ul>
)}
</div>
<div className="card" style={{ marginTop: "1rem" }}>
<h3 style={{ marginTop: 0 }}>Historie</h3>
{akte.historie.length === 0 && <p className="text-muted">Noch keine Änderungen protokolliert.</p>}
{akte.historie.length > 0 && (
<ul>
{akte.historie.map((h) => (
<li key={h.id}>
{new Date(h.zeitpunkt).toLocaleString("de-DE")} {h.ereignistyp}
{h.benutzer_name && <span className="text-muted"> ({h.benutzer_name})</span>}
</li>
))}
</ul>
)}
</div>
<div style={{ marginTop: "1rem" }}>
<Link to={`/objekte/${objekt.id}/kontrolle`} className="btn btn-primary">
Kontrolle starten
</Link>
</div>
</main>
);
}
+8
View File
@@ -82,6 +82,14 @@ export function ObjektListPage() {
: "Noch nie kontrolliert"}
</div>
</Link>
<Link
to={`/akte/objekt/${objekt.id}`}
className="text-muted"
style={{ fontSize: "0.85rem" }}
onClick={(e) => e.stopPropagation()}
>
Akte ansehen
</Link>
</li>
))}
</ul>