Materialänderungslog: Historie-Endpunkt + Admin-Tab

historie-Tabelle (append-only Audit-Trail über Kontrolle/Fehlbestand/
Nachfüllung/Mindermenge) war bisher nur intern befüllt, ohne API/UI. Neu:
GET /api/v1/historie (gefiltert nach Typ/ID/Benutzer/Zeitraum, paginiert,
Benutzername aufgelöst), Admin-Portal-Tab "Änderungslog" mit Typ-Filter.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV

backend/DEVLOG.md

@@ -105,3 +105,14 @@ Keine Commits in dieser Session.
 - backend/scripts/test_label_erzeugen.py | 34 ++++++++++++++++++++++++++++++++++
 
 ---
+## 2026-09-04 19:52 – 19:53 (1m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- ...07_indizes_und_unique_constraints.py => 0007_indizes_und_locks.py} | 4 ++--
+
+---

backend/app/api/v1/api.py

@@ -7,6 +7,7 @@ from app.api.v1.endpoints import (
     eskalation,
     fehlbestaende,
     health,
+    historie,
     kontrollen,
     objekte,
     stammdaten,
@@ -25,4 +26,5 @@ api_router.include_router(objekte.router, tags=["objekte"])
 api_router.include_router(kontrollen.router, tags=["kontrollen"])
 api_router.include_router(fehlbestaende.router, tags=["fehlbestaende"])
 api_router.include_router(eskalation.router, tags=["eskalation"])
+api_router.include_router(historie.router, tags=["historie"])
 api_router.include_router(dashboard.router, tags=["dashboard"])

backend/app/api/v1/endpoints/historie.py

@@ -0,0 +1,63 @@
+from datetime import datetime
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import require_roles
+from app.db.session import get_db
+from app.models.auth import Benutzer, RolleTyp
+from app.models.historie import Historie
+from app.schemas.historie import HistorieRead
+
+router = APIRouter()
+
+_verantwortliche = require_roles(
+    RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
+)
+
+
+@router.get("/historie", response_model=list[HistorieRead])
+async def liste_historie(
+    entitaet_typ: str | None = None,
+    entitaet_id: str | None = None,
+    benutzer_id: int | None = None,
+    von: datetime | None = None,
+    bis: datetime | None = None,
+    limit: int = 100,
+    offset: int = 0,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_verantwortliche),
+) -> list[HistorieRead]:
+    """Materialänderungslog (Prompt 13): append-only Audit-Trail über Kontrolle,
+    Fehlbestand, Nachfüllung, Mindermenge. Neueste zuerst, Standard-Seitengröße
+    100 (openapi-Review-Fund: Listen-Endpunkte brauchen Pagination)."""
+    stmt = select(Historie, Benutzer.name).outerjoin(Benutzer, Benutzer.id == Historie.benutzer_id)
+    if entitaet_typ is not None:
+        stmt = stmt.where(Historie.entitaet_typ == entitaet_typ)
+    if entitaet_id is not None:
+        stmt = stmt.where(Historie.entitaet_id == entitaet_id)
+    if benutzer_id is not None:
+        stmt = stmt.where(Historie.benutzer_id == benutzer_id)
+    if von is not None:
+        stmt = stmt.where(Historie.zeitpunkt >= von)
+    if bis is not None:
+        stmt = stmt.where(Historie.zeitpunkt <= bis)
+    stmt = stmt.order_by(Historie.zeitpunkt.desc()).limit(min(limit, 500)).offset(offset)
+
+    result = await db.execute(stmt)
+    return [
+        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 result.all()
+    ]

backend/app/schemas/historie.py

@@ -0,0 +1,18 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class HistorieRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    zeitpunkt: datetime
+    benutzer_id: int | None
+    benutzer_name: str | None
+    ereignistyp: str
+    entitaet_typ: str
+    entitaet_id: str
+    alter_wert: dict | None
+    neuer_wert: dict | None
+    begruendung: str | None

backend/tests/test_historie.py

@@ -60,3 +60,40 @@ async def test_historie_kette_soll10_ist8_genehmigt_dann_nachgefuellt(
     # Reine Teilfolge-Prüfung (keine anderen Events dazwischen erwartet, aber falls
     # doch, ist die Reihenfolge der genannten Ereignisse das eigentlich Wichtige).
     assert ereignisse == erwartete_reihenfolge
+
+
+@pytest.mark.asyncio
+async def test_historie_endpunkt_liefert_gefiltert(
+    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
+    )
+    kontrolle_id = start.json()["id"]
+    await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": "10"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.get(
+        "/api/v1/historie",
+        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
+        headers=auth_header(verantwortlicher_token),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert len(body) >= 1
+    assert all(e["entitaet_typ"] == "kontrolle" for e in body)
+    assert body[0]["benutzer_name"] == "Test Mitarbeiter"
+
+
+@pytest.mark.asyncio
+async def test_historie_mitarbeiter_ohne_zugriff(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.get("/api/v1/historie", headers=auth_header(token))
+    assert response.status_code == 403