feat(personal): Personal/Qualifikationen-Modul (Roadmap Phase 1)

Neues Modul für die BOS-Ressourcenplattform-Erweiterung: Einheit (Zug/Gruppe,
self-referenzierend), Qualifikationstyp (Führerschein/Lehrgang/Berechtigung),
BenutzerQualifikation (mit Gültigkeit) und ObjekttypQualifikationsanforderung
(M:N) - beantwortet "wer darf dieses Fahrzeug fahren?" über
GET /objekte/{id}/berechtigung/{benutzer_id}.

Benutzer und Objekt bekommen optionale einheit_id (additiv, analog
fahrzeug_id-Muster). Nebenbei Bugfix: PATCH /benutzer konnte einheit_id nicht
auf null setzen (Feld-vorhanden-Check via model_fields_set statt is not None).

Frontend: neuer Admin-Tab "Personal" (Einheiten, Qualifikationstypen,
Benutzer-Qualifikationszuordnung), Einheit-Auswahl im Benutzer-Formular.

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

backend/alembic/versions/0012_personal_qualifikationen.py

@@ -0,0 +1,86 @@
+"""Personal/Qualifikationen-Modul (Roadmap-Erweiterung, Phase 1 nach BOS-
+Ressourcenplattform-Anforderung): Einheit-Hierarchie, Qualifikationstypen
+(Führerschein/Lehrgang/Berechtigung), Benutzer-Qualifikationszuordnung mit
+Gültigkeit, sowie welche Qualifikation ein Objekttyp (z.B. Fahrzeug) zum
+Bedienen voraussetzt - beantwortet "wer darf dieses Fahrzeug fahren?".
+
+Revision ID: 0012_personal_qualifikationen
+Revises: 0011_objekttyp_zugfahrzeug
+Create Date: 2026-09-05
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0012_personal_qualifikationen"
+down_revision: Union[str, None] = "0011_objekttyp_zugfahrzeug"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("CREATE TYPE qualifikationskategorie AS ENUM ('fuehrerschein', 'lehrgang', 'berechtigung')")
+
+    op.execute(
+        """
+        CREATE TABLE einheit (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL,
+            uebergeordnete_einheit_id INTEGER REFERENCES einheit(id),
+            standort_id INTEGER REFERENCES standort(id),
+            CONSTRAINT einheit_kein_selbstbezug CHECK (uebergeordnete_einheit_id <> id)
+        )
+        """
+    )
+    op.execute("CREATE INDEX idx_einheit_uebergeordnet ON einheit (uebergeordnete_einheit_id)")
+
+    op.execute("ALTER TABLE benutzer ADD COLUMN einheit_id INTEGER REFERENCES einheit(id)")
+    # Fahrzeuge/Objekte künftig einer Einheit zuordenbar (analog fahrzeug_id-Muster
+    # aus Migration 0010: additiv, NULL = keiner Einheit zugeordnet).
+    op.execute("ALTER TABLE objekt ADD COLUMN einheit_id INTEGER REFERENCES einheit(id)")
+
+    op.execute(
+        """
+        CREATE TABLE qualifikationstyp (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL,
+            kategorie qualifikationskategorie NOT NULL,
+            gueltigkeit_monate INTEGER
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE benutzer_qualifikation (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            benutzer_id INTEGER NOT NULL REFERENCES benutzer(id),
+            qualifikationstyp_id INTEGER NOT NULL REFERENCES qualifikationstyp(id),
+            erworben_am DATE NOT NULL,
+            gueltig_bis DATE,
+            nachweis_nummer TEXT,
+            UNIQUE (benutzer_id, qualifikationstyp_id, erworben_am)
+        )
+        """
+    )
+    op.execute("CREATE INDEX idx_benutzer_qualifikation_benutzer ON benutzer_qualifikation (benutzer_id)")
+
+    op.execute(
+        """
+        CREATE TABLE objekttyp_qualifikationsanforderung (
+            objekttyp_id INTEGER NOT NULL REFERENCES objekttyp(id),
+            qualifikationstyp_id INTEGER NOT NULL REFERENCES qualifikationstyp(id),
+            PRIMARY KEY (objekttyp_id, qualifikationstyp_id)
+        )
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE objekttyp_qualifikationsanforderung")
+    op.execute("DROP TABLE benutzer_qualifikation")
+    op.execute("DROP TABLE qualifikationstyp")
+    op.execute("ALTER TABLE objekt DROP COLUMN einheit_id")
+    op.execute("ALTER TABLE benutzer DROP COLUMN einheit_id")
+    op.execute("DROP TABLE einheit")
+    op.execute("DROP TYPE IF EXISTS qualifikationskategorie")

backend/app/api/v1/api.py

@@ -11,6 +11,7 @@ from app.api.v1.endpoints import (
     historie,
     kontrollen,
     objekte,
+    personal,
     stammdaten,
     vorlagen,
     zustaendigkeit,
@@ -30,3 +31,4 @@ api_router.include_router(eskalation.router, tags=["eskalation"])
 api_router.include_router(historie.router, tags=["historie"])
 api_router.include_router(geraet_instanz.router, tags=["geraet_instanz"])
 api_router.include_router(dashboard.router, tags=["dashboard"])
+api_router.include_router(personal.router, tags=["personal"])

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

@@ -37,6 +37,7 @@ async def erstelle_benutzer(
         email=payload.email,
         passwort_hash=hash_password(payload.passwort),
         aktiv=True,
+        einheit_id=payload.einheit_id,
     )
     db.add(benutzer)
     await db.flush()
@@ -64,6 +65,11 @@ async def aendere_benutzer(
         benutzer.email = payload.email
     if payload.aktiv is not None:
         benutzer.aktiv = payload.aktiv
+    if "einheit_id" in payload.model_fields_set:
+        # explizit gesetzt (auch null = Einheit lösen) unterscheiden von "nicht
+        # 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 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

@@ -0,0 +1,213 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_current_user, require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.models.personal import (
+    BenutzerQualifikation,
+    Einheit,
+    ObjekttypQualifikationsanforderung,
+    Qualifikationstyp,
+)
+from app.schemas.personal import (
+    BenutzerQualifikationCreate,
+    BenutzerQualifikationRead,
+    BerechtigungspruefungRead,
+    EinheitCreate,
+    EinheitRead,
+    EinheitUpdate,
+    ObjekttypQualifikationsanforderungCreate,
+    ObjekttypQualifikationsanforderungRead,
+    QualifikationstypCreate,
+    QualifikationstypRead,
+    QualifikationstypUpdate,
+)
+from app.services.personal import pruefe_berechtigung
+
+router = APIRouter()
+
+_admin_only = require_roles(RolleTyp.administration)
+
+
+# -- Einheit ------------------------------------------------------------------
+
+@router.get("/einheiten", response_model=list[EinheitRead])
+async def liste_einheiten(db: AsyncSession = Depends(get_db), _=Depends(get_current_user)) -> list[Einheit]:
+    result = await db.execute(select(Einheit))
+    return list(result.scalars().all())
+
+
+@router.post("/einheiten", response_model=EinheitRead, status_code=status.HTTP_201_CREATED)
+async def erstelle_einheit(
+    payload: EinheitCreate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Einheit:
+    einheit = Einheit(**payload.model_dump())
+    db.add(einheit)
+    try:
+        await db.flush()
+    except IntegrityError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Ungültige Zuordnung") from exc
+    return einheit
+
+
+@router.patch("/einheiten/{einheit_id}", response_model=EinheitRead)
+async def aendere_einheit(
+    einheit_id: int, payload: EinheitUpdate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Einheit:
+    einheit = await db.get(Einheit, einheit_id)
+    if einheit is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Einheit nicht gefunden")
+    for feld, wert in payload.model_dump(exclude_unset=True).items():
+        setattr(einheit, 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 einheit
+
+
+# -- Qualifikationstyp ---------------------------------------------------------
+
+@router.get("/qualifikationstypen", response_model=list[QualifikationstypRead])
+async def liste_qualifikationstypen(
+    db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
+) -> list[Qualifikationstyp]:
+    result = await db.execute(select(Qualifikationstyp))
+    return list(result.scalars().all())
+
+
+@router.post("/qualifikationstypen", response_model=QualifikationstypRead, status_code=status.HTTP_201_CREATED)
+async def erstelle_qualifikationstyp(
+    payload: QualifikationstypCreate, db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> Qualifikationstyp:
+    qualifikationstyp = Qualifikationstyp(**payload.model_dump())
+    db.add(qualifikationstyp)
+    await db.flush()
+    return qualifikationstyp
+
+
+@router.patch("/qualifikationstypen/{qualifikationstyp_id}", response_model=QualifikationstypRead)
+async def aendere_qualifikationstyp(
+    qualifikationstyp_id: int,
+    payload: QualifikationstypUpdate,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> Qualifikationstyp:
+    qualifikationstyp = await db.get(Qualifikationstyp, qualifikationstyp_id)
+    if qualifikationstyp is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Qualifikationstyp nicht gefunden")
+    for feld, wert in payload.model_dump(exclude_unset=True).items():
+        setattr(qualifikationstyp, feld, wert)
+    await db.flush()
+    return qualifikationstyp
+
+
+# -- Benutzer-Qualifikation -----------------------------------------------------
+
+@router.get("/benutzer/{benutzer_id}/qualifikationen", response_model=list[BenutzerQualifikationRead])
+async def liste_benutzer_qualifikationen(
+    benutzer_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
+) -> list[BenutzerQualifikation]:
+    result = await db.execute(
+        select(BenutzerQualifikation).where(BenutzerQualifikation.benutzer_id == benutzer_id)
+    )
+    return list(result.scalars().all())
+
+
+@router.post(
+    "/benutzer/{benutzer_id}/qualifikationen",
+    response_model=BenutzerQualifikationRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def erfasse_benutzer_qualifikation(
+    benutzer_id: int,
+    payload: BenutzerQualifikationCreate,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> BenutzerQualifikation:
+    if payload.benutzer_id != benutzer_id:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="benutzer_id im Pfad und Body weichen ab")
+    qualifikation = BenutzerQualifikation(**payload.model_dump())
+    db.add(qualifikation)
+    try:
+        await db.flush()
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Qualifikation mit diesem Erwerbsdatum existiert bereits"
+        ) from exc
+    return qualifikation
+
+
+# -- Objekttyp-Qualifikationsanforderung -----------------------------------------
+
+@router.get(
+    "/objekttypen/{objekttyp_id}/qualifikationsanforderungen",
+    response_model=list[ObjekttypQualifikationsanforderungRead],
+)
+async def liste_qualifikationsanforderungen(
+    objekttyp_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
+) -> list[ObjekttypQualifikationsanforderung]:
+    result = await db.execute(
+        select(ObjekttypQualifikationsanforderung).where(
+            ObjekttypQualifikationsanforderung.objekttyp_id == objekttyp_id
+        )
+    )
+    return list(result.scalars().all())
+
+
+@router.post(
+    "/objekttypen/{objekttyp_id}/qualifikationsanforderungen",
+    response_model=ObjekttypQualifikationsanforderungRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def erstelle_qualifikationsanforderung(
+    objekttyp_id: int,
+    payload: ObjekttypQualifikationsanforderungCreate,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> ObjekttypQualifikationsanforderung:
+    if payload.objekttyp_id != objekttyp_id:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="objekttyp_id im Pfad und Body weichen ab")
+    anforderung = ObjekttypQualifikationsanforderung(**payload.model_dump())
+    db.add(anforderung)
+    try:
+        await db.flush()
+    except IntegrityError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Anforderung existiert bereits") from exc
+    return anforderung
+
+
+@router.delete(
+    "/objekttypen/{objekttyp_id}/qualifikationsanforderungen/{qualifikationstyp_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+)
+async def loesche_qualifikationsanforderung(
+    objekttyp_id: int,
+    qualifikationstyp_id: int,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> None:
+    anforderung = await db.get(
+        ObjekttypQualifikationsanforderung, {"objekttyp_id": objekttyp_id, "qualifikationstyp_id": qualifikationstyp_id}
+    )
+    if anforderung is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Anforderung nicht gefunden")
+    await db.delete(anforderung)
+    await db.flush()
+
+
+# -- Berechtigungsprüfung -------------------------------------------------------
+
+@router.get("/objekte/{objekt_id}/berechtigung/{benutzer_id}", response_model=BerechtigungspruefungRead)
+async def pruefe_objekt_berechtigung(
+    objekt_id: int, benutzer_id: int, db: AsyncSession = Depends(get_db), _=Depends(get_current_user)
+) -> BerechtigungspruefungRead:
+    """Nutzer-Beispiel Personal-Modul: "wer darf dieses Fahrzeug fahren?"."""
+    fehlende = await pruefe_berechtigung(db, benutzer_id=benutzer_id, objekt_id=objekt_id)
+    return BerechtigungspruefungRead(
+        berechtigt=len(fehlende) == 0,
+        fehlende_qualifikationstypen=list(fehlende),
+    )

backend/app/models/auth.py

@@ -44,6 +44,7 @@ class Benutzer(Base):
     passwort_hash: Mapped[str] = mapped_column(String, nullable=False)
     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)
 
     rollen: Mapped[list["BenutzerRolle"]] = relationship(back_populates="benutzer", lazy="selectin")
 

backend/app/models/objekt.py

@@ -40,3 +40,6 @@ class Objekt(Base):
     fahrzeug_id: Mapped[int | None] = mapped_column(
         ForeignKey("objekt.id", ondelete="SET NULL")
     )
+    # Personal-Modul (Migration 0012): Zuordnung zu einer Organisationseinheit
+    # (Zug/Gruppe), additiv/optional wie fahrzeug_id.
+    einheit_id: Mapped[int | None] = mapped_column(ForeignKey("einheit.id"))

backend/app/models/personal.py

@@ -0,0 +1,86 @@
+import enum
+import uuid
+from datetime import date
+
+from sqlalchemy import CheckConstraint, Date, ForeignKey, String, UniqueConstraint
+from sqlalchemy.dialects.postgresql import ENUM as PgEnum, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class Qualifikationskategorie(str, enum.Enum):
+    fuehrerschein = "fuehrerschein"
+    lehrgang = "lehrgang"
+    berechtigung = "berechtigung"
+
+
+qualifikationskategorie_pg = PgEnum(
+    Qualifikationskategorie, name="qualifikationskategorie", create_type=False
+)
+
+
+class Einheit(Base):
+    """Roadmap-Erweiterung (Personal-Modul): Organisationseinheit (Zug/Gruppe),
+    self-referenzierend für Hierarchie (analog Kategorie.ueberkategorie_id).
+    Bewusst eigenständig statt an Standort gekoppelt - eine Einheit kann
+    ortsübergreifend sein, ein Standort mehrere Einheiten beherbergen."""
+
+    __tablename__ = "einheit"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    uebergeordnete_einheit_id: Mapped[int | None] = mapped_column(ForeignKey("einheit.id"))
+    standort_id: Mapped[int | None] = mapped_column(ForeignKey("standort.id"))
+
+    __table_args__ = (
+        CheckConstraint("uebergeordnete_einheit_id <> id", name="einheit_kein_selbstbezug"),
+    )
+
+
+class Qualifikationstyp(Base):
+    """Führerschein/Lehrgang/Berechtigung als ein Stammdaten-Konzept, damit
+    "welche Qualifikation braucht Objekttyp X" generisch bleibt (kein separates
+    Führerschein-Feld am Fahrzeug nötig, Karte-Nachtrag Personal-Modul)."""
+
+    __tablename__ = "qualifikationstyp"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    kategorie: Mapped[Qualifikationskategorie] = mapped_column(
+        qualifikationskategorie_pg, nullable=False
+    )
+    # NULL = unbefristet (z.B. manche Berechtigungen), sonst Gültigkeitsdauer in
+    # Monaten ab erworben_am - Ablauf wird nicht hier, sondern je
+    # BenutzerQualifikation über gueltig_bis geführt (individuelles Datum je
+    # Nachweis, nicht rein rechnerisch aus dem Typ abgeleitet).
+    gueltigkeit_monate: Mapped[int | None] = mapped_column()
+
+
+class BenutzerQualifikation(Base):
+    __tablename__ = "benutzer_qualifikation"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    benutzer_id: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
+    qualifikationstyp_id: Mapped[int] = mapped_column(ForeignKey("qualifikationstyp.id"), nullable=False)
+    erworben_am: Mapped[date] = mapped_column(Date, nullable=False)
+    gueltig_bis: Mapped[date | None] = mapped_column(Date)
+    nachweis_nummer: Mapped[str | None] = mapped_column(String)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "benutzer_id", "qualifikationstyp_id", "erworben_am", name="uq_benutzer_qualifikation"
+        ),
+    )
+
+
+class ObjekttypQualifikationsanforderung(Base):
+    """M:N: welche Qualifikationstypen braucht man, um Objekte dieses Objekttyps
+    zu bedienen/fahren (Nutzer-Beispiel: "wer darf GW-San 01 fahren?")."""
+
+    __tablename__ = "objekttyp_qualifikationsanforderung"
+
+    objekttyp_id: Mapped[int] = mapped_column(ForeignKey("objekttyp.id"), primary_key=True)
+    qualifikationstyp_id: Mapped[int] = mapped_column(
+        ForeignKey("qualifikationstyp.id"), primary_key=True
+    )

backend/app/schemas/benutzer.py

@@ -9,6 +9,7 @@ class BenutzerCreate(BaseModel):
     passwort: str
     email: str | None = None
     rollen: list[RolleTyp] = []
+    einheit_id: int | None = None
 
 
 class BenutzerRead(BaseModel):
@@ -19,6 +20,7 @@ class BenutzerRead(BaseModel):
     email: str | None
     aktiv: bool
     rollen: list[str]
+    einheit_id: int | None
 
     @classmethod
     def from_orm_benutzer(cls, benutzer) -> "BenutzerRead":
@@ -29,6 +31,7 @@ class BenutzerRead(BaseModel):
             email=benutzer.email,
             aktiv=benutzer.aktiv,
             rollen=benutzer.rollen_namen,
+            einheit_id=benutzer.einheit_id,
         )
 
 
@@ -37,3 +40,4 @@ class BenutzerUpdate(BaseModel):
     email: str | None = None
     aktiv: bool | None = None
     rollen: list[RolleTyp] | None = None
+    einheit_id: int | None = None

backend/app/schemas/personal.py

@@ -0,0 +1,82 @@
+import uuid
+from datetime import date
+
+from pydantic import BaseModel, ConfigDict
+
+from app.models.personal import Qualifikationskategorie
+
+
+class EinheitCreate(BaseModel):
+    name: str
+    uebergeordnete_einheit_id: int | None = None
+    standort_id: int | None = None
+
+
+class EinheitRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    name: str
+    uebergeordnete_einheit_id: int | None
+    standort_id: int | None
+
+
+class EinheitUpdate(BaseModel):
+    name: str | None = None
+    uebergeordnete_einheit_id: int | None = None
+    standort_id: int | None = None
+
+
+class QualifikationstypCreate(BaseModel):
+    name: str
+    kategorie: Qualifikationskategorie
+    gueltigkeit_monate: int | None = None
+
+
+class QualifikationstypRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    name: str
+    kategorie: Qualifikationskategorie
+    gueltigkeit_monate: int | None
+
+
+class QualifikationstypUpdate(BaseModel):
+    name: str | None = None
+    kategorie: Qualifikationskategorie | None = None
+    gueltigkeit_monate: int | None = None
+
+
+class BenutzerQualifikationCreate(BaseModel):
+    benutzer_id: int
+    qualifikationstyp_id: int
+    erworben_am: date
+    gueltig_bis: date | None = None
+    nachweis_nummer: str | None = None
+
+
+class BenutzerQualifikationRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    benutzer_id: int
+    qualifikationstyp_id: int
+    erworben_am: date
+    gueltig_bis: date | None
+    nachweis_nummer: str | None
+
+
+class ObjekttypQualifikationsanforderungCreate(BaseModel):
+    objekttyp_id: int
+    qualifikationstyp_id: int
+
+
+class ObjekttypQualifikationsanforderungRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    objekttyp_id: int
+    qualifikationstyp_id: int
+
+
+class BerechtigungspruefungRead(BaseModel):
+    """Antwort auf "darf Benutzer X Objekt Y bedienen/fahren?" (Nutzer-Beispiel)."""
+
+    berechtigt: bool
+    fehlende_qualifikationstypen: list[QualifikationstypRead]

backend/app/services/personal.py

@@ -0,0 +1,53 @@
+from datetime import date
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.objekt import Objekt
+from app.models.personal import (
+    BenutzerQualifikation,
+    ObjekttypQualifikationsanforderung,
+    Qualifikationstyp,
+)
+
+
+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
+    Benutzer nicht (mehr) gültig nachgewiesen hat - leere Liste = berechtigt."""
+    objekt = await db.get(Objekt, objekt_id)
+    if objekt is None:
+        return []
+
+    erforderlich = (
+        (
+            await db.execute(
+                select(Qualifikationstyp)
+                .join(
+                    ObjekttypQualifikationsanforderung,
+                    ObjekttypQualifikationsanforderung.qualifikationstyp_id == Qualifikationstyp.id,
+                )
+                .where(ObjekttypQualifikationsanforderung.objekttyp_id == objekt.objekttyp_id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    if not erforderlich:
+        return []
+
+    heute = date.today()
+    gueltige_ids = set(
+        (
+            await db.execute(
+                select(BenutzerQualifikation.qualifikationstyp_id).where(
+                    BenutzerQualifikation.benutzer_id == benutzer_id,
+                    (BenutzerQualifikation.gueltig_bis.is_(None))
+                    | (BenutzerQualifikation.gueltig_bis >= heute),
+                )
+            )
+        )
+        .scalars()
+        .all()
+    )
+    return [q for q in erforderlich if q.id not in gueltige_ids]

backend/tests/test_personal.py

@@ -0,0 +1,139 @@
+from datetime import date, timedelta
+
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_einheit_anlegen_und_hierarchie(client, admin_user):
+    token = await login(client, "admin1")
+    zug = await client.post("/api/v1/einheiten", json={"name": "Sanitätszug 1"}, headers=auth_header(token))
+    assert zug.status_code == 201
+    zug_id = zug.json()["id"]
+
+    gruppe = await client.post(
+        "/api/v1/einheiten",
+        json={"name": "Gruppe 1", "uebergeordnete_einheit_id": zug_id},
+        headers=auth_header(token),
+    )
+    assert gruppe.status_code == 201
+    assert gruppe.json()["uebergeordnete_einheit_id"] == zug_id
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_einheit_nicht_anlegen(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.post("/api/v1/einheiten", json={"name": "X"}, headers=auth_header(token))
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_qualifikationstyp_anlegen(client, admin_user):
+    token = await login(client, "admin1")
+    response = await client.post(
+        "/api/v1/qualifikationstypen",
+        json={"name": "Führerschein C1", "kategorie": "fuehrerschein"},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 201
+    assert response.json()["gueltigkeit_monate"] is None
+
+
+@pytest.mark.asyncio
+async def test_wer_darf_fahrzeug_fahren(
+    client, db_session, objekttyp_factory, standort_factory, admin_user, mitarbeiter_user, hauptserver_id
+):
+    """Nutzer-Beispiel Personal-Modul: Objekttyp verlangt Führerschein C1,
+    Benutzer ohne gültigen Nachweis darf nicht, mit gültigem Nachweis schon."""
+    from app.services.vorlagen import erstelle_vorlage
+    from app.services.objekte import objekt_aus_vorlage
+
+    objekttyp = await objekttyp_factory("GW-San", ist_zugfahrzeug=True)
+    standort = await standort_factory("Wache-Personal")
+    vorlage = await erstelle_vorlage(db_session, objekttyp_id=objekttyp.id, name="GW-San Standard", positionen=[])
+    fahrzeug = await objekt_aus_vorlage(
+        db_session,
+        code="GW-SAN-01",
+        name="GW-San 01",
+        standort_id=standort.id,
+        vorlage=vorlage,
+        zustaendiger_server_id=hauptserver_id,
+    )
+
+    token = await login(client, "admin1")
+    quali = await client.post(
+        "/api/v1/qualifikationstypen",
+        json={"name": "Führerschein C1", "kategorie": "fuehrerschein"},
+        headers=auth_header(token),
+    )
+    quali_id = quali.json()["id"]
+
+    await client.post(
+        f"/api/v1/objekttypen/{objekttyp.id}/qualifikationsanforderungen",
+        json={"objekttyp_id": objekttyp.id, "qualifikationstyp_id": quali_id},
+        headers=auth_header(token),
+    )
+
+    mitarbeiter_id = mitarbeiter_user.id
+
+    # ohne Nachweis: nicht berechtigt
+    pruefung = await client.get(
+        f"/api/v1/objekte/{fahrzeug.id}/berechtigung/{mitarbeiter_id}", headers=auth_header(token)
+    )
+    assert pruefung.json()["berechtigt"] is False
+    assert len(pruefung.json()["fehlende_qualifikationstypen"]) == 1
+
+    # abgelaufener Nachweis zählt nicht
+    await client.post(
+        f"/api/v1/benutzer/{mitarbeiter_id}/qualifikationen",
+        json={
+            "benutzer_id": mitarbeiter_id,
+            "qualifikationstyp_id": quali_id,
+            "erworben_am": (date.today() - timedelta(days=1000)).isoformat(),
+            "gueltig_bis": (date.today() - timedelta(days=1)).isoformat(),
+        },
+        headers=auth_header(token),
+    )
+    abgelaufen = await client.get(
+        f"/api/v1/objekte/{fahrzeug.id}/berechtigung/{mitarbeiter_id}", headers=auth_header(token)
+    )
+    assert abgelaufen.json()["berechtigt"] is False
+
+    # gültiger Nachweis: berechtigt
+    await client.post(
+        f"/api/v1/benutzer/{mitarbeiter_id}/qualifikationen",
+        json={
+            "benutzer_id": mitarbeiter_id,
+            "qualifikationstyp_id": quali_id,
+            "erworben_am": date.today().isoformat(),
+            "gueltig_bis": None,
+        },
+        headers=auth_header(token),
+    )
+    berechtigt = await client.get(
+        f"/api/v1/objekte/{fahrzeug.id}/berechtigung/{mitarbeiter_id}", headers=auth_header(token)
+    )
+    assert berechtigt.json()["berechtigt"] is True
+    assert berechtigt.json()["fehlende_qualifikationstypen"] == []
+
+
+@pytest.mark.asyncio
+async def test_benutzer_einheit_zuordnung(client, admin_user):
+    token = await login(client, "admin1")
+    einheit = await client.post("/api/v1/einheiten", json={"name": "Zug 2"}, headers=auth_header(token))
+    einheit_id = einheit.json()["id"]
+
+    benutzer = await client.post(
+        "/api/v1/benutzer",
+        json={"name": "Neuer Helfer", "login": "helfer1", "passwort": "test-passwort-123", "einheit_id": einheit_id},
+        headers=auth_header(token),
+    )
+    assert benutzer.status_code == 201
+    assert benutzer.json()["einheit_id"] == einheit_id
+
+    benutzer_id = benutzer.json()["id"]
+    geloest = await client.patch(
+        f"/api/v1/benutzer/{benutzer_id}", json={"einheit_id": None}, headers=auth_header(token)
+    )
+    assert geloest.json()["einheit_id"] is None