Fahrzeug/Anhänger-Kopplung (objekt.gezogen_von_objekt_id)

Nutzer-Vorgabe: Fahrzeug und Anhänger beide als eigenständige Objekte
möglich, optional fest gekoppelt. Selbstreferenzierende nullable FK, kein
eigener Anhänger-Objekttyp nötig (Objekt bleibt generisch, Karte 11).
POST /objekte übergibt das Feld, neuer PATCH /objekte/{id} zum
nachträglichen Setzen/Lösen.

Von postgres-expert gegengeprüft, Fixes übernommen: ON DELETE SET NULL
(Zugfahrzeug entfernen blockiert Anhänger nicht), CHECK-Constraint gegen
Selbstbezug, Existenzprüfung + direkter Zyklus-Check (A↔B) im Endpoint statt
unhandled 500. Mehrstufige Ketten bewusst nicht geprüft (kein Anwendungsfall).

Tests: Anlage mit/ohne Kopplung, nachträglich setzen/lösen, Selbstbezug,
ungültiges Ziel, direkter Zyklus.

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

backend/DEVLOG.md

@@ -189,3 +189,39 @@ Keine Commits in dieser Session.
 - frontend/src/pages/admin/ZustaendigkeitSection.tsx | 194 +++++++++++++++++++++
 
 ---
+## 2026-09-04 20:43 – 20:46 (3m)
+**Beschreibung:** Claude Code Session
+**Projekt:** backend
+
+### Commits
+- d61aaba Karte 14 Schritt 1: geraet_instanz-Tabelle (additiv)
+
+### Geänderte Dateien
+- arbeitskarten/00_index.md                       |  1 +
+- arbeitskarten/14_pruef_wartungspflicht.md       | 98 ++++++++++++++++++++++++
+- backend/DEVLOG.md                               | 42 +++++++++++
+- backend/alembic/versions/0008_geraet_instanz.py | 65 ++++++++++++++++
+- backend/app/models/__init__.py                  |  3 +
+- backend/app/models/geraet_instanz.py            | 40 ++++++++++
+- backend/app/models/objektposition.py            |  6 +-
+- backend/tests/test_geraet_instanz.py            | 99 +++++++++++++++++++++++++
+
+---
+## 2026-09-04 20:47 – 20:47 (0m)
+**Beschreibung:** Claude Code Session
+**Projekt:** backend
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- arbeitskarten/00_index.md                       |  1 +
+- arbeitskarten/14_pruef_wartungspflicht.md       | 98 ++++++++++++++++++++++++
+- backend/DEVLOG.md                               | 42 +++++++++++
+- backend/alembic/versions/0008_geraet_instanz.py | 65 ++++++++++++++++
+- backend/app/models/__init__.py                  |  3 +
+- backend/app/models/geraet_instanz.py            | 40 ++++++++++
+- backend/app/models/objektposition.py            |  6 +-
+- backend/tests/test_geraet_instanz.py            | 99 +++++++++++++++++++++++++
+
+---

backend/alembic/versions/0009_gezogen_von.py

@@ -0,0 +1,42 @@
+"""Fahrzeug/Anhänger-Kopplung: objekt.gezogen_von_objekt_id (additiv)
+
+Revision ID: 0009_gezogen_von
+Revises: 0008_geraet_instanz
+Create Date: 2026-09-04
+
+Nutzer-Vorgabe: Fahrzeuge und Anhänger sollen beide als eigenständige Objekte
+gehen (unabhängig, kein Zwang) UND optional fest gekoppelt werden können
+(Anhänger gehört zu einem Zugfahrzeug). Selbstreferenzierende, nullable FK -
+kein eigener Anhänger-Objekttyp/-Tabelle nötig, Objekt bleibt generisch
+(Karte 11).
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0009_gezogen_von"
+down_revision: Union[str, None] = "0008_geraet_instanz"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "ALTER TABLE objekt ADD COLUMN gezogen_von_objekt_id INTEGER "
+        "REFERENCES objekt(id) ON DELETE SET NULL"
+    )
+    op.execute("CREATE INDEX idx_objekt_gezogen_von ON objekt (gezogen_von_objekt_id)")
+    # postgres-expert-Review-Fund: DB-Sicherheitsnetz gegen Selbstbezug (Objekt
+    # zieht sich selbst) - Anwendungscode prüft zusätzlich direkte Zyklen (A↔B),
+    # mehrstufige Ketten sind fachlich nicht vorgesehen und bewusst nicht per DB
+    # erzwungen (Aufwand/Nutzen).
+    op.execute(
+        "ALTER TABLE objekt ADD CONSTRAINT objekt_kein_selbstbezug "
+        "CHECK (gezogen_von_objekt_id IS DISTINCT FROM id)"
+    )
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE objekt DROP CONSTRAINT objekt_kein_selbstbezug")
+    op.execute("DROP INDEX idx_objekt_gezogen_von")
+    op.execute("ALTER TABLE objekt DROP COLUMN gezogen_von_objekt_id")

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

@@ -20,6 +20,7 @@ from app.schemas.objekt import (
     ObjektpositionRead,
     ObjektpositionUpdate,
     ObjektRead,
+    ObjektUpdate,
 )
 from app.services.objekte import (
     dupliziere_objekt,
@@ -192,6 +193,9 @@ async def erstelle_objekt(
     vorlage = await db.get(Beladungsvorlage, payload.vorlage_id)
     if vorlage is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Vorlage nicht gefunden")
+    if payload.gezogen_von_objekt_id is not None:
+        if await db.get(Objekt, payload.gezogen_von_objekt_id) is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Zugfahrzeug nicht gefunden")
 
     hauptserver = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
     zustaendiger_server_id = hauptserver.scalar_one()
@@ -204,11 +208,45 @@ async def erstelle_objekt(
             standort_id=payload.standort_id,
             vorlage=vorlage,
             zustaendiger_server_id=zustaendiger_server_id,
+            gezogen_von_objekt_id=payload.gezogen_von_objekt_id,
         )
     except IntegrityError as exc:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Code bereits vergeben") from exc
 
 
+@router.patch("/objekte/{objekt_id}", response_model=ObjektRead)
+async def aendere_objekt(
+    objekt_id: int, payload: ObjektUpdate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Objekt:
+    """Fahrzeug/Anhänger-Kopplung nachträglich setzen/lösen (Nutzer-Vorgabe).
+    postgres-expert-Review-Fund: Zielobjekt-Existenz und direkter Zyklus (A↔B)
+    werden hier geprüft, mehrstufige Ketten sind fachlich nicht vorgesehen und
+    bewusst nicht geprüft (Aufwand/Nutzen)."""
+    objekt = await db.get(Objekt, objekt_id)
+    if objekt is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+
+    daten = payload.model_dump(exclude_unset=True)
+    ziel_id = daten.get("gezogen_von_objekt_id")
+    if ziel_id is not None:
+        ziel = await db.get(Objekt, ziel_id)
+        if ziel is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Zugfahrzeug nicht gefunden")
+        if ziel.gezogen_von_objekt_id == objekt_id:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail="Zyklus: Zielobjekt wird bereits von diesem Objekt gezogen",
+            )
+
+    for feld, wert in daten.items():
+        setattr(objekt, feld, wert)
+    try:
+        await db.flush()
+    except IntegrityError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Ungültige Zuordnung") from exc
+    return objekt
+
+
 @router.post(
     "/objekte/{objekt_id}/duplizieren", response_model=ObjektRead, status_code=status.HTTP_201_CREATED
 )

backend/app/models/objekt.py

@@ -33,3 +33,9 @@ class Objekt(Base):
     zustaendiger_server_id: Mapped[int] = mapped_column(
         ForeignKey("systemknoten.id"), nullable=False
     )
+    # Fahrzeug/Anhänger-Kopplung (Nutzer-Vorgabe): optional, beide Fälle möglich -
+    # Anhänger als eigenständiges Objekt (NULL) oder fest einem Zugfahrzeug
+    # zugeordnet. Selbstreferenzierend, kein eigener Anhänger-Objekttyp nötig.
+    gezogen_von_objekt_id: Mapped[int | None] = mapped_column(
+        ForeignKey("objekt.id", ondelete="SET NULL")
+    )

backend/app/schemas/objekt.py

@@ -14,6 +14,7 @@ class ObjektCreate(BaseModel):
     objekttyp_id: int
     vorlage_id: int
     standort_id: int
+    gezogen_von_objekt_id: int | None = None
 
 
 class ObjektRead(BaseModel):
@@ -25,6 +26,11 @@ class ObjektRead(BaseModel):
     vorlage_id: int | None
     standort_id: int
     status: ObjektStatus
+    gezogen_von_objekt_id: int | None
+
+
+class ObjektUpdate(BaseModel):
+    gezogen_von_objekt_id: int | None = None
 
 
 class ObjektDuplizierenRequest(BaseModel):

backend/app/services/objekte.py

@@ -56,9 +56,12 @@ async def objekt_aus_vorlage(
     standort_id: int,
     vorlage: Beladungsvorlage,
     zustaendiger_server_id: int,
+    gezogen_von_objekt_id: int | None = None,
 ) -> Objekt:
     """Prompt 09 (Neuanlage, nicht Duplizieren): Objektpositionen werden aus den
-    Vorlagenpositionen initialisiert, Ist-Menge startet bei 0 ("nicht kontrolliert")."""
+    Vorlagenpositionen initialisiert, Ist-Menge startet bei 0 ("nicht kontrolliert").
+    gezogen_von_objekt_id: optionale Fahrzeug/Anhänger-Kopplung (Nutzer-Vorgabe),
+    beide Fälle möglich - unabhängiges Objekt (NULL) oder fest zugeordnet."""
     objekt = Objekt(
         code=code,
         name=name,
@@ -66,6 +69,7 @@ async def objekt_aus_vorlage(
         vorlage_id=vorlage.id,
         standort_id=standort_id,
         zustaendiger_server_id=zustaendiger_server_id,
+        gezogen_von_objekt_id=gezogen_von_objekt_id,
     )
     db.add(objekt)
     await db.flush()

backend/tests/test_objekte.py

@@ -298,3 +298,134 @@ async def test_objektposition_label_pdf_ohne_code_ist_404(
     mit_code = await client.get(f"/api/v1/objektpositionen/{position_id}/label.pdf", headers=auth_header(token))
     assert mit_code.status_code == 200
     assert mit_code.content.startswith(b"%PDF")
+
+
+@pytest.mark.asyncio
+async def test_anhaenger_kopplung_bei_anlage_und_nachtraeglich(
+    client, db_session, objekttyp_factory, standort_factory, admin_user
+):
+    """Nutzer-Vorgabe: Fahrzeug/Anhänger beide als eigenständige Objekte möglich,
+    optional per gezogen_von_objekt_id gekoppelt."""
+    objekttyp = await objekttyp_factory("Fahrzeug/Anhänger")
+    standort = await standort_factory("Wache-Kopplung")
+    vorlage = await erstelle_vorlage(
+        db_session, objekttyp_id=objekttyp.id, name="Kopplung Standard", positionen=[]
+    )
+    token = await login(client, "admin1")
+
+    fahrzeug = await client.post(
+        "/api/v1/objekte",
+        json={
+            "code": "FZ-K1",
+            "name": "Zugfahrzeug",
+            "objekttyp_id": objekttyp.id,
+            "vorlage_id": vorlage.id,
+            "standort_id": standort.id,
+        },
+        headers=auth_header(token),
+    )
+    fahrzeug_id = fahrzeug.json()["id"]
+    assert fahrzeug.json()["gezogen_von_objekt_id"] is None
+
+    anhaenger = await client.post(
+        "/api/v1/objekte",
+        json={
+            "code": "AH-K1",
+            "name": "Anhänger",
+            "objekttyp_id": objekttyp.id,
+            "vorlage_id": vorlage.id,
+            "standort_id": standort.id,
+            "gezogen_von_objekt_id": fahrzeug_id,
+        },
+        headers=auth_header(token),
+    )
+    assert anhaenger.status_code == 201
+    assert anhaenger.json()["gezogen_von_objekt_id"] == fahrzeug_id
+
+    # unabhängiger Anhänger, erst nachträglich gekoppelt
+    unabhaengig = await client.post(
+        "/api/v1/objekte",
+        json={
+            "code": "AH-K2",
+            "name": "Anhänger 2",
+            "objekttyp_id": objekttyp.id,
+            "vorlage_id": vorlage.id,
+            "standort_id": standort.id,
+        },
+        headers=auth_header(token),
+    )
+    unabhaengig_id = unabhaengig.json()["id"]
+    assert unabhaengig.json()["gezogen_von_objekt_id"] is None
+
+    geaendert = await client.patch(
+        f"/api/v1/objekte/{unabhaengig_id}",
+        json={"gezogen_von_objekt_id": fahrzeug_id},
+        headers=auth_header(token),
+    )
+    assert geaendert.status_code == 200
+    assert geaendert.json()["gezogen_von_objekt_id"] == fahrzeug_id
+
+    geloest = await client.patch(
+        f"/api/v1/objekte/{unabhaengig_id}",
+        json={"gezogen_von_objekt_id": None},
+        headers=auth_header(token),
+    )
+    assert geloest.json()["gezogen_von_objekt_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_anhaenger_kopplung_lehnt_selbstbezug_und_zyklus_ab(
+    client, db_session, objekttyp_factory, standort_factory, admin_user
+):
+    objekttyp = await objekttyp_factory("Fahrzeug/Anhänger Zyklus")
+    standort = await standort_factory("Wache-Zyklus")
+    vorlage = await erstelle_vorlage(
+        db_session, objekttyp_id=objekttyp.id, name="Zyklus Standard", positionen=[]
+    )
+    token = await login(client, "admin1")
+
+    a = await client.post(
+        "/api/v1/objekte",
+        json={
+            "code": "ZY-A",
+            "name": "Objekt A",
+            "objekttyp_id": objekttyp.id,
+            "vorlage_id": vorlage.id,
+            "standort_id": standort.id,
+        },
+        headers=auth_header(token),
+    )
+    a_id = a.json()["id"]
+
+    # Selbstbezug: A zieht sich selbst
+    selbstbezug = await client.patch(
+        f"/api/v1/objekte/{a_id}", json={"gezogen_von_objekt_id": a_id}, headers=auth_header(token)
+    )
+    assert selbstbezug.status_code == 409
+
+    # nicht existierendes Zielobjekt
+    ungueltig = await client.patch(
+        f"/api/v1/objekte/{a_id}", json={"gezogen_von_objekt_id": 999999}, headers=auth_header(token)
+    )
+    assert ungueltig.status_code == 404
+
+    b = await client.post(
+        "/api/v1/objekte",
+        json={
+            "code": "ZY-B",
+            "name": "Objekt B",
+            "objekttyp_id": objekttyp.id,
+            "vorlage_id": vorlage.id,
+            "standort_id": standort.id,
+            "gezogen_von_objekt_id": a_id,
+        },
+        headers=auth_header(token),
+    )
+    b_id = b.json()["id"]
+    assert b.json()["gezogen_von_objekt_id"] == a_id
+
+    # Zyklus: A soll jetzt von B gezogen werden (B wird bereits von A gezogen)
+    zyklus = await client.patch(
+        f"/api/v1/objekte/{a_id}", json={"gezogen_von_objekt_id": b_id}, headers=auth_header(token)
+    )
+    assert zyklus.status_code == 409