feat(personal): PERS-004 Einsatzfunktionen (Gruppenführer/Melder/...)

Neue Stammdaten-Tabelle einsatzfunktion (Migration 0031), benutzer bekommt
optionales einsatzfunktion_id-Feld - analog zu einheit_id modelliert, eine
Person hat im Regelfall genau eine primäre Einsatzfunktion. Bewusst
getrennt von den technischen App-Rollen (RolleTyp), die nur steuern, was
jemand in MABEA tun darf, nicht welche Funktion er im Einsatz hat.

CRUD-Endpunkt /einsatzfunktionen nach Kategorie-Muster, Duplikat-Prüfung
vorab (Projekt-Konvention). Admin-UI: Auswahl + Inline-Neuanlage in der
Benutzerverwaltung (BenutzerSection).

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

backend/alembic/versions/0031_einsatzfunktion.py

@@ -0,0 +1,33 @@
+"""PERS-004: Einsatzfunktionen (Gruppenführer/Melder/...), benutzer.einsatzfunktion_id.
+
+Revision ID: 0031_einsatzfunktion
+Revises: 0030_hersteller_modell
+Create Date: 2026-09-08
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0031_einsatzfunktion"
+down_revision: Union[str, None] = "0030_hersteller_modell"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE einsatzfunktion (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL UNIQUE
+        )
+        """
+    )
+    op.execute("ALTER TABLE benutzer ADD COLUMN einsatzfunktion_id INTEGER REFERENCES einsatzfunktion(id)")
+    op.execute("CREATE INDEX ix_benutzer_einsatzfunktion_id ON benutzer (einsatzfunktion_id)")
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_benutzer_einsatzfunktion_id")
+    op.execute("ALTER TABLE benutzer DROP COLUMN einsatzfunktion_id")
+    op.execute("DROP TABLE einsatzfunktion")

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

@@ -38,6 +38,7 @@ async def erstelle_benutzer(
         passwort_hash=hash_password(payload.passwort),
         aktiv=True,
         einheit_id=payload.einheit_id,
+        einsatzfunktion_id=payload.einsatzfunktion_id,
     )
     db.add(benutzer)
     await db.flush()
@@ -70,6 +71,8 @@ async def aendere_benutzer(
         # im Request enthalten" - anders als die übrigen Felder hier, da null
         # ein gültiger, gewollter Zielzustand ist (Bugfix vor Auslieferung).
         benutzer.einheit_id = payload.einheit_id
+    if "einsatzfunktion_id" in payload.model_fields_set:
+        benutzer.einsatzfunktion_id = payload.einsatzfunktion_id
     if payload.rollen is not None:
         result = await db.execute(select(BenutzerRolle).where(BenutzerRolle.benutzer_id == benutzer_id))
         for bestehende in result.scalars().all():

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

@@ -9,6 +9,7 @@ from app.models.auth import KnotenTyp, RolleTyp, Systemknoten
 from app.models.personal import (
     BenutzerQualifikation,
     Einheit,
+    Einsatzfunktion,
     ObjekttypQualifikationsanforderung,
     Qualifikationstyp,
 )
@@ -19,6 +20,8 @@ from app.schemas.personal import (
     EinheitCreate,
     EinheitRead,
     EinheitUpdate,
+    EinsatzfunktionCreate,
+    EinsatzfunktionRead,
     ObjekttypQualifikationsanforderungCreate,
     ObjekttypQualifikationsanforderungRead,
     QualifikationstypCreate,
@@ -75,6 +78,31 @@ async def aendere_einheit(
     return einheit
 
 
+# -- Einsatzfunktion (PERS-004) ------------------------------------------------
+
+@router.get("/einsatzfunktionen", response_model=list[EinsatzfunktionRead])
+async def liste_einsatzfunktionen(
+    db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
+) -> list[Einsatzfunktion]:
+    result = await db.execute(select(Einsatzfunktion))
+    return list(result.scalars().all())
+
+
+@router.post("/einsatzfunktionen", response_model=EinsatzfunktionRead, status_code=status.HTTP_201_CREATED)
+async def erstelle_einsatzfunktion(
+    payload: EinsatzfunktionCreate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Einsatzfunktion:
+    # Explizit vorab prüfen statt sich auf den UNIQUE-Constraint zu verlassen
+    # (PendingRollbackError-Risiko in geteilter Session, siehe geraet_instanz.py).
+    bestehende = await db.execute(select(Einsatzfunktion).where(Einsatzfunktion.name == payload.name))
+    if bestehende.scalar_one_or_none() is not None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Einsatzfunktion existiert bereits")
+    einsatzfunktion = Einsatzfunktion(**payload.model_dump())
+    db.add(einsatzfunktion)
+    await db.flush()
+    return einsatzfunktion
+
+
 # -- Qualifikationstyp ---------------------------------------------------------
 
 @router.get("/qualifikationstypen", response_model=list[QualifikationstypRead])

backend/app/models/auth.py

@@ -45,6 +45,8 @@ class Benutzer(Base):
     aktiv: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
     email: Mapped[str | None] = mapped_column(String)  # Sprint 6 / Karte 05 (Migration 0003)
     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"))
 
     rollen: Mapped[list["BenutzerRolle"]] = relationship(back_populates="benutzer", lazy="selectin")
 

backend/app/models/personal.py

@@ -74,6 +74,21 @@ class BenutzerQualifikation(Base):
     )
 
 
+class Einsatzfunktion(Base):
+    """PERS-004: Funktion im Einsatzkontext (z.B. Gruppenführer, Truppführer,
+    Melder, Maschinist, Atemschutzgeräteträger) - unabhängig von den
+    technischen App-Rollen (RolleTyp: mitarbeiter/materialverantwortlicher/
+    ...), die nur steuern, was jemand in MABEA tun darf, nicht welche Funktion
+    er im Einsatz hat. Bewusst analog zu Benutzer.einheit_id als einzelnes
+    optionales Feld modelliert, nicht als M:N - eine Person hat im Regelfall
+    genau eine primäre Einsatzfunktion, keine Historie/Zeitraum nötig."""
+
+    __tablename__ = "einsatzfunktion"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, unique=True, nullable=False)
+
+
 class ObjekttypQualifikationsanforderung(Base):
     """M:N: welche Qualifikationstypen braucht man, um Objekte dieses Objekttyps
     zu bedienen/fahren (Nutzer-Beispiel: "wer darf GW-San 01 fahren?")."""

backend/app/schemas/benutzer.py

@@ -10,6 +10,7 @@ class BenutzerCreate(BaseModel):
     email: str | None = None
     rollen: list[RolleTyp] = []
     einheit_id: int | None = None
+    einsatzfunktion_id: int | None = None
 
 
 class BenutzerRead(BaseModel):
@@ -21,6 +22,7 @@ class BenutzerRead(BaseModel):
     aktiv: bool
     rollen: list[str]
     einheit_id: int | None
+    einsatzfunktion_id: int | None
 
     @classmethod
     def from_orm_benutzer(cls, benutzer) -> "BenutzerRead":
@@ -32,6 +34,7 @@ class BenutzerRead(BaseModel):
             aktiv=benutzer.aktiv,
             rollen=benutzer.rollen_namen,
             einheit_id=benutzer.einheit_id,
+            einsatzfunktion_id=benutzer.einsatzfunktion_id,
         )
 
 
@@ -41,3 +44,4 @@ class BenutzerUpdate(BaseModel):
     aktiv: bool | None = None
     rollen: list[RolleTyp] | None = None
     einheit_id: int | None = None
+    einsatzfunktion_id: int | None = None

backend/app/schemas/personal.py

@@ -26,6 +26,15 @@ class EinheitUpdate(BaseModel):
     standort_id: int | None = None
 
 
+class EinsatzfunktionCreate(BaseModel):
+    name: str
+
+
+class EinsatzfunktionRead(EinsatzfunktionCreate):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+
+
 class QualifikationstypCreate(BaseModel):
     name: str
     kategorie: Qualifikationskategorie

backend/tests/test_personal.py

@@ -139,6 +139,51 @@ async def test_benutzer_einheit_zuordnung(client, admin_user):
     assert geloest.json()["einheit_id"] is None
 
 
+@pytest.mark.asyncio
+async def test_einsatzfunktion_anlegen_und_benutzer_zuordnen(client, admin_user):
+    """PERS-004: Einsatzfunktion unabhängig von der technischen App-Rolle."""
+    token = await login(client, "admin1")
+
+    funktion = await client.post(
+        "/api/v1/einsatzfunktionen", json={"name": "Gruppenführer"}, headers=auth_header(token)
+    )
+    assert funktion.status_code == 201
+    funktion_id = funktion.json()["id"]
+
+    duplikat = await client.post(
+        "/api/v1/einsatzfunktionen", json={"name": "Gruppenführer"}, headers=auth_header(token)
+    )
+    assert duplikat.status_code == 409
+
+    benutzer = await client.post(
+        "/api/v1/benutzer",
+        json={
+            "name": "Neuer Gruppenführer",
+            "login": "gf1",
+            "passwort": "test-passwort-123",
+            "einsatzfunktion_id": funktion_id,
+        },
+        headers=auth_header(token),
+    )
+    assert benutzer.status_code == 201
+    assert benutzer.json()["einsatzfunktion_id"] == funktion_id
+
+    benutzer_id = benutzer.json()["id"]
+    geloest = await client.patch(
+        f"/api/v1/benutzer/{benutzer_id}", json={"einsatzfunktion_id": None}, headers=auth_header(token)
+    )
+    assert geloest.json()["einsatzfunktion_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_einsatzfunktion_nicht_anlegen(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.post(
+        "/api/v1/einsatzfunktionen", json={"name": "Melder"}, headers=auth_header(token)
+    )
+    assert response.status_code == 403
+
+
 @pytest.mark.asyncio
 async def test_bald_ablaufende_qualifikation_erscheint_im_dashboard(
     client, admin_user, mitarbeiter_user