feat(personal): PERS-001/002 Person fachlich von Benutzer getrennt

Neue Tabelle person (Migration 0034, Backfill bestehender Benutzer 1:1
per Zeilennummer, keine Datenverluste). Benutzer.person_id optional -
Login kann bestehende Person verknüpfen oder legt automatisch eine neue
an (Rückwärtskompatibilität zum bisherigen Anlegen-Flow). /personen-CRUD
(admin-only), neue PersonSection.tsx im Personal-Tab, BenutzerSection
um Personen-Auswahl erweitert.

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

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

@@ -6,6 +6,7 @@ from app.api.deps import require_roles
 from app.core.security import hash_password
 from app.db.session import get_db
 from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
+from app.models.personal import Person
 from app.schemas.benutzer import BenutzerCreate, BenutzerRead, BenutzerUpdate
 
 router = APIRouter()
@@ -31,6 +32,19 @@ async def erstelle_benutzer(
     if result.scalar_one_or_none() is not None:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Login bereits vergeben")
 
+    person_id = payload.person_id
+    if person_id is not None:
+        if await db.get(Person, person_id) is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Person nicht gefunden")
+    else:
+        # PERS-002: kein bestehender Person-Datensatz gewählt -> automatisch
+        # anlegen, damit jeder Benutzer weiterhin ohne Zusatzschritt eine
+        # verknüpfte Person hat (Rückwärtskompatibilität zum bisherigen Flow).
+        person = Person(name=payload.name)
+        db.add(person)
+        await db.flush()
+        person_id = person.id
+
     benutzer = Benutzer(
         name=payload.name,
         login=payload.login,
@@ -39,6 +53,7 @@ async def erstelle_benutzer(
         aktiv=True,
         einheit_id=payload.einheit_id,
         einsatzfunktion_id=payload.einsatzfunktion_id,
+        person_id=person_id,
     )
     db.add(benutzer)
     await db.flush()

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

@@ -11,6 +11,7 @@ from app.models.personal import (
     Einheit,
     Einsatzfunktion,
     ObjekttypQualifikationsanforderung,
+    Person,
     Qualifikationstyp,
 )
 from app.schemas.personal import (
@@ -24,12 +25,15 @@ from app.schemas.personal import (
     EinsatzfunktionRead,
     ObjekttypQualifikationsanforderungCreate,
     ObjekttypQualifikationsanforderungRead,
+    PersonCreate,
+    PersonRead,
+    PersonUpdate,
     QualifikationstypCreate,
     QualifikationstypRead,
     QualifikationstypUpdate,
 )
 from app.services import historie as historie_service
-from app.services.personal import pruefe_berechtigung
+from app.services.personal import erstelle_person, liste_personen, pruefe_berechtigung
 
 router = APIRouter()
 
@@ -41,6 +45,39 @@ async def _hauptserver_id(db: AsyncSession) -> int:
 _admin_only = require_roles(RolleTyp.administration)
 
 
+# -- Person (PERS-001) ---------------------------------------------------------
+
+@router.get("/personen", response_model=list[PersonRead])
+async def liste_personen_endpunkt(db: AsyncSession = Depends(get_db), _=Depends(_admin_only)) -> list[Person]:
+    return await liste_personen(db)
+
+
+@router.post("/personen", response_model=PersonRead, status_code=status.HTTP_201_CREATED)
+async def erstelle_person_endpunkt(
+    payload: PersonCreate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Person:
+    return await erstelle_person(
+        db, name=payload.name, geburtsdatum=payload.geburtsdatum, kontakt=payload.kontakt
+    )
+
+
+@router.patch("/personen/{person_id}", response_model=PersonRead)
+async def aendere_person(
+    person_id: int, payload: PersonUpdate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Person:
+    person = await db.get(Person, person_id)
+    if person is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Person nicht gefunden")
+    if payload.name is not None:
+        person.name = payload.name
+    if "geburtsdatum" in payload.model_fields_set:
+        person.geburtsdatum = payload.geburtsdatum
+    if "kontakt" in payload.model_fields_set:
+        person.kontakt = payload.kontakt
+    await db.flush()
+    return person
+
+
 # -- Einheit ------------------------------------------------------------------
 
 @router.get("/einheiten", response_model=list[EinheitRead])

backend/app/models/auth.py

@@ -47,6 +47,11 @@ class Benutzer(Base):
     einheit_id: Mapped[int | None] = mapped_column(ForeignKey("einheit.id"))  # Personal-Modul (Migration 0012)
     # PERS-004: primäre Funktion im Einsatzkontext, optional.
     einsatzfunktion_id: Mapped[int | None] = mapped_column(ForeignKey("einsatzfunktion.id"))
+    # PERS-002: Verknüpfung zum fachlichen Person-Datensatz. Bestehende Benutzer
+    # wurden per Migration 0034 rückwirkend mit einer Person verknüpft, bleibt
+    # aber nullable - ein Login ohne fachliche Person(Übergang) wäre sonst nicht
+    # anlegbar, wenn die Migration künftig übersprungen wird.
+    person_id: Mapped[int | None] = mapped_column(ForeignKey("person.id"))
 
     rollen: Mapped[list["BenutzerRolle"]] = relationship(back_populates="benutzer", lazy="selectin")
 

backend/app/models/personal.py

@@ -20,6 +20,19 @@ qualifikationskategorie_pg = PgEnum(
 )
 
 
+class Person(Base):
+    """PERS-001: fachliches Konzept einer Person, getrennt vom technischen Login
+    (Benutzer). Kann ohne Benutzerkonto existieren (z.B. Jugendgruppe, externe
+    Helfer ohne Systemzugang) - Benutzer.person_id verweist optional hierher."""
+
+    __tablename__ = "person"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    geburtsdatum: Mapped[date | None] = mapped_column(Date)
+    kontakt: Mapped[str | None] = mapped_column(String)
+
+
 class Einheit(Base):
     """Roadmap-Erweiterung (Personal-Modul): Organisationseinheit (Zug/Gruppe),
     self-referenzierend für Hierarchie (analog Kategorie.ueberkategorie_id).

backend/app/schemas/benutzer.py

@@ -11,6 +11,10 @@ class BenutzerCreate(BaseModel):
     rollen: list[RolleTyp] = []
     einheit_id: int | None = None
     einsatzfunktion_id: int | None = None
+    # PERS-002: bestehende Person verknüpfen statt automatisch eine neue anzulegen
+    # (z.B. wenn eine bereits erfasste Person - Jugendgruppe o.ä. - nachträglich
+    # einen Login bekommt).
+    person_id: int | None = None
 
 
 class BenutzerRead(BaseModel):
@@ -23,6 +27,7 @@ class BenutzerRead(BaseModel):
     rollen: list[str]
     einheit_id: int | None
     einsatzfunktion_id: int | None
+    person_id: int | None
 
     @classmethod
     def from_orm_benutzer(cls, benutzer) -> "BenutzerRead":
@@ -35,6 +40,7 @@ class BenutzerRead(BaseModel):
             rollen=benutzer.rollen_namen,
             einheit_id=benutzer.einheit_id,
             einsatzfunktion_id=benutzer.einsatzfunktion_id,
+            person_id=benutzer.person_id,
         )
 
 

backend/app/schemas/personal.py

@@ -6,6 +6,26 @@ from pydantic import BaseModel, ConfigDict
 from app.models.personal import Qualifikationskategorie
 
 
+class PersonCreate(BaseModel):
+    name: str
+    geburtsdatum: date | None = None
+    kontakt: str | None = None
+
+
+class PersonRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    name: str
+    geburtsdatum: date | None
+    kontakt: str | None
+
+
+class PersonUpdate(BaseModel):
+    name: str | None = None
+    geburtsdatum: date | None = None
+    kontakt: str | None = None
+
+
 class EinheitCreate(BaseModel):
     name: str
     uebergeordnete_einheit_id: int | None = None

backend/app/services/personal.py

@@ -7,10 +7,23 @@ from app.models.objekt import Objekt
 from app.models.personal import (
     BenutzerQualifikation,
     ObjekttypQualifikationsanforderung,
+    Person,
     Qualifikationstyp,
 )
 
 
+async def erstelle_person(db: AsyncSession, *, name: str, geburtsdatum, kontakt: str | None) -> Person:
+    person = Person(name=name, geburtsdatum=geburtsdatum, kontakt=kontakt)
+    db.add(person)
+    await db.flush()
+    return person
+
+
+async def liste_personen(db: AsyncSession) -> list[Person]:
+    result = await db.execute(select(Person))
+    return list(result.scalars().all())
+
+
 async def pruefe_berechtigung(db: AsyncSession, *, benutzer_id: int, objekt_id: int) -> list[Qualifikationstyp]:
     """Nutzer-Beispiel Personal-Modul: "wer darf dieses Fahrzeug fahren?".
     Liefert die Qualifikationstypen, die der Objekttyp verlangt, aber der