Was ist das User Management?

Das User Management ist die zentrale Verwaltungsplattform für Division Mayhem. Es verbindet euer Authentik-Konto mit Discord, TeamSpeak 3 und Steam und sorgt dafür, dass ihr auf allen Plattformen automatisch die richtigen Rollen und Gruppen bekommt.

Du meldest dich mit deinen bestehenden Clan-Zugangsdaten (Authentik) an — ein separates Passwort gibt es nicht.

Wie melde ich mich an?

Die Anmeldung läuft vollständig über Authentik — kein separates Konto, kein zweites Passwort.

Was sehe ich auf meinem Dashboard?

Das Dashboard zeigt dein aktuelles Profil mit allen Daten, die das System über dich kennt. Du erreichst es über Profil in der Navigation oder über dein Benutzermenü oben rechts.

Um Discord ID, TS3 UID und Steam ID zu hinterlegen, klicke auf dem Dashboard auf Profil bearbeiten.

Wo finde ich meine IDs?

Damit Discord- und TeamSpeak-Synchronisation funktioniert, musst du deine jeweilige ID einmalig im Profil hinterlegen.

Was passiert nach einer Gruppenänderung?

Sobald dein Rang oder deine Gruppenmitgliedschaften durch die Mitgliederverwaltung geändert werden, synchronisiert das System automatisch im Hintergrund:

Die Synchronisation startet direkt nach dem Speichern — du musst nichts weiter tun. Sollten Rollen nach einigen Minuten noch nicht stimmen, kann die Mitgliederverwaltung den Sync manuell anstoßen.

Mitglieder und Gruppen verwalten

Mitglieder der Gruppe Mitgliederverwaltung haben Zugriff auf die Benutzerliste und Gruppenübersicht.

Benutzer bearbeiten

Zusätzliche Admin-Funktionen

Etwas funktioniert nicht?

Komponenten & Technologien

Verzeichnisstruktur

backend/app/
├── config.py               # Settings (pydantic-settings, .env)
├── main.py                 # FastAPI-App, Routen für HTML-Seiten
├── database.py             # SQLAlchemy async engine + session
├── middleware/
│   └── auth.py             # JWT decode, Gruppen-Cache, Depends
├── models/
│   └── attribute_option.py # SQLAlchemy ORM-Modell
├── routers/
│   ├── auth.py             # OIDC Login-Flow, /me, /logout
│   ├── users.py            # Benutzer CRUD + auto-sync trigger
│   ├── groups.py           # Gruppen CRUD (nur Authentik)
│   ├── attribute_options.py# Positionen CRUD (lokale DB)
│   ├── discord.py          # Discord OAuth2 + Sync
│   └── teamspeak.py        # TS3 ServerQuery Sync
├── schemas/                # Pydantic request/response models
└── services/
    ├── authentik_api.py    # Authentik REST API v3 Client
    ├── auth.py             # JWT erstellen / dekodieren
    ├── discord.py          # Discord Bot + OAuth2 API
    └── teamspeak.py        # TS3 ServerQuery Client

Benutzer und Gruppen existieren ausschließlich in Authentik. Die lokale PostgreSQL-Datenbank speichert nur die Pickup-Tabellen für Benutzerattribute.

Settings-Klasse & .env

Alle Konfiguration läuft über app/config.py — eine pydantic-settings BaseSettings-Klasse, die automatisch Umgebungsvariablen und .env liest. Das @lru_cache() auf get_settings() stellt sicher, dass die Datei nur einmal beim Start geparst wird:

class Settings(BaseSettings):
    secret_key: str = "change-this-in-production"
    algorithm:  str = "HS256"
    access_token_expire_minutes: int = 60

    authentik_url:    str = ""   # http://authentik:9000
    authentik_token:  str = ""   # Service-Account-Token (Admin → Tokens)
    oidc_client_id:   str = ""
    oidc_client_secret: str = ""
    oidc_redirect_uri:  str = ""

    um_access_group: str = "Mitgliederverwaltung"
    um_admin_group:  str = "UM-Administrators"

    discord_bot_token: str = ""
    discord_guild_id:  str = ""

    ts3_host:      str = ""
    ts3_port:      int = 10011
    ts3_username:  str = "serveradmin"
    ts3_password:  str = ""
    ts3_server_id: int = 1

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Authorization Code Flow & JWT-Ausstellung

Datei: routers/auth.py + services/auth.py. Der gesamte Login läuft ohne Passwort-Handling im UM — Authentik ist alleinige Identitätsquelle.

Schritt 1 — GET /api/auth/oidc/login

state = secrets.token_urlsafe(32)   # CSRF-Schutz
params = {
    "response_type": "code",
    "client_id":     s.oidc_client_id,
    "redirect_uri":  _redirect_uri(),
    "scope":         "openid email profile",
    "state":         state,
}
# httpOnly Cookie "oidc_state" für 600 Sekunden setzen
redirect.set_cookie("oidc_state", state, httponly=True, samesite="lax", max_age=600)
# Browser → https://{AUTHENTIK_URL}/application/o/authorize/?{params}

Schritt 2 — GET /api/auth/oidc/callback?code=…&state=…

# 1. State validieren
if state != request.cookies.get("oidc_state"):
    return RedirectResponse("/login?error=state_mismatch")

# 2. Code gegen Token tauschen
POST {AUTHENTIK_URL}/application/o/token/
  grant_type=authorization_code, code, redirect_uri, client_id, client_secret
→ {"access_token": "…", "id_token": "…", …}

# 3. Benutzerinfo abrufen
GET {AUTHENTIK_URL}/application/o/userinfo/
  Authorization: Bearer {access_token}
→ {"sub": "uuid", "preferred_username": "…", "email": "…"}

# 4. Benutzer in Authentik nachschlagen (nach Username)
ak_user = await ak.get_user_by_username(username)
#   → GET /api/v3/core/users/?username={name}&page_size=1

# 5. Zugang prüfen (direkte Gruppenmitgliedschaft via groups_obj)
direct_groups = {g["name"] for g in ak_user.groups_obj}
is_admin  = UM_ADMIN_GROUP  in direct_groups
is_member = UM_ACCESS_GROUP in direct_groups or is_admin
if not is_member: return RedirectResponse("/login?error=access_denied")

# 6. Discord-ID auto-befüllen (best-effort)
if not ak_user.discord_id:
    discord_id = await ak.get_discord_id_from_source(ak_user.pk)
    if discord_id:
        await ak.update_user_attributes(ak_user.pk, {"um_discord_id": discord_id})

Schritt 3 — JWT ausstellen (services/auth.py)

payload = {
    "sub":      uid,       # Authentik UUID des Benutzers
    "ak_pk":   ak_user.pk, # Authentik Integer-PK (für API-Calls)
    "username": username,
    "email":    email,
    "is_admin": is_admin,
    "exp":      utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
}
token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")  # python-jose

# Sicher als httpOnly Cookie setzen (kein JS-Zugriff)
response.set_cookie("access_token", token,
    httponly=True, samesite="lax",
    max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60)

Jede Anfrage: JWT → Gruppen-Check → CurrentUser

Datei: middleware/auth.py. Jeder geschützte Endpunkt hat Depends(get_current_user) oder Depends(require_admin).

Token-Extraktion & Dekodierung

async def get_current_user(request, credentials=Depends(bearer_scheme)):
    # 1. Token aus Bearer-Header ODER httpOnly-Cookie lesen
    token = credentials.credentials if credentials \
            else request.cookies.get("access_token")
    if not token:
        raise HTTPException(401, "Not authenticated")

    # 2. JWT dekodieren (python-jose, HS256, SECRET_KEY)
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    # Fehlschlag → None → 401

    # 3. CurrentUser aus Claims bauen
    user = CurrentUser(
        id=payload["sub"], ak_pk=int(payload["ak_pk"]),
        username=payload["username"], email=payload["email"],
        is_admin=bool(payload.get("is_admin", False)),
    )
    return await _verify_groups(user)  # live Gruppen-Check

Gruppen-Cache (_verify_groups)

# In-memory Cache: ak_pk → (is_member, is_admin, checked_at)
_group_cache: dict[int, tuple[bool, bool, float]] = {}
GROUP_CHECK_TTL = 300  # 5 Minuten

async def _verify_groups(user: CurrentUser) -> CurrentUser:
    now = time.monotonic()
    cached = _group_cache.get(user.ak_pk)

    if cached:
        is_member, is_admin, ts = cached
        if now - ts < GROUP_CHECK_TTL:
            if not is_member:
                raise HTTPException(401, "Zugriff verweigert")
            user.is_admin = is_admin
            return user   # Cache-Hit: kein Authentik-Call nötig

    # Cache-Miss: Live-Check gegen Authentik
    try:
        ak_user = await ak.get_user_by_pk(user.ak_pk)  # Detail-Endpoint
        direct_groups = {g["name"] for g in ak_user.groups_obj}
        is_admin  = UM_ADMIN_GROUP  in direct_groups
        is_member = UM_ACCESS_GROUP in direct_groups or is_admin
        _group_cache[user.ak_pk] = (is_member, is_admin, now)
    except Exception as exc:
        logger.warning("group-check FAILED, JWT-Fallback: %s", exc)
        return user   # Fallback: JWT-Wert wird behalten

    if not is_member:
        raise HTTPException(401, "Nicht Mitglied der Mitgliederverwaltung")
    user.is_admin = is_admin
    return user

def invalidate_group_cache(ak_pk: int) -> None:
    _group_cache.pop(ak_pk, None)  # sofortiger Re-Check bei nächster Anfrage

Permission Tiers

API-Client, Attribute lesen & schreiben

API-Client (services/authentik_api.py)

Alle Aufrufe an Authentik gehen über einen zentralen async HTTP-Client auf Basis von httpx. Jeder Call öffnet einen eigenen Client-Kontext. Die Basis-URL und der Token werden aus den Settings gezogen:

def _headers() -> dict:
    return {
        "Authorization": f"Bearer {get_settings().authentik_token}",
        "Content-Type": "application/json",
    }

def _base() -> str:
    return get_settings().authentik_url.rstrip("/") + "/api/v3"

Alle Management-Calls nutzen den Service-Account-Token (AUTHENTIK_TOKEN aus .env). Der Token wird im Authentik Admin Interface unter Directory → Tokens angelegt und benötigt Admin-Rechte.

Attribute lesen

Authentik liefert bei GET /api/v3/core/users/{pk}/ ein JSON-Objekt mit einem freien attributes-Dict. Das UM liest daraus alle eigenen Felder direkt über Python-Properties auf der AuthentikUser-Klasse:

class AuthentikUser:
    def __init__(self, data: dict) -> None:
        self.pk         = data["pk"]           # int — Primärschlüssel
        self.uid        = data.get("uid", "")  # str — UUID
        self.attributes = data.get("attributes") or {}
        self.groups_obj = data.get("groups_obj", [])  # [{pk, name, ...}]

    @property
    def discord_id(self) -> str | None:
        return self.attributes.get("um_discord_id")

    @property
    def teamspeak_id(self) -> str | None:
        return self.attributes.get("um_teamspeak_id")

    @property
    def clan_rank(self) -> str | None:
        return self.attributes.get("um_clan_rank")

Der Detail-Endpoint /core/users/{pk}/ wird bewusst bevorzugt — der List-Endpoint mit ?pk=-Filter ist unzuverlässig bei Integer-PKs und liefert mitunter leere oder falsche Ergebnisse (bekannter Authentik-Bug).

Attribute schreiben — Read → Merge → PATCH

Authentik's PATCH /core/users/{pk}/ ersetzt das gesamte attributes-Dict wenn es mitgeschickt wird. Ein einfaches PATCH mit nur dem neuen Wert würde alle anderen Attribute löschen. Deshalb nutzt das UM ein Read-Modify-Write-Muster:

async def update_user_attributes(pk: int, attributes: dict) -> AuthentikUser:
    # 1. Aktuellen Stand lesen
    user = await get_user_by_pk(pk)

    # 2. Neue Werte in bestehendes Dict mergen
    merged = {**user.attributes, **attributes}
    #         ^^ alle alten Keys   ^^ neue Keys überschreiben / ergänzen

    # 3. Komplettes Dict zurückschreiben
    return await update_user(pk, attributes=merged)

Jeder Aufruf in den Routers sieht so aus:

# Profil speichern (users.py)
attrs = {}
for field in ("discord_id", "teamspeak_id", "steam_id"):
    val = getattr(data, field)
    if val is not None:
        attrs[f"um_{field}"] = val   # Präfix "um_" wird hier gesetzt

await ak.update_user_attributes(user_pk, attrs)
# → liest erst alle Attribute, merged, schreibt komplett zurück

Gruppenattribute schreiben

Dasselbe Merge-Muster gilt für Gruppen. update_group() trennt zuerst Top-Level-Felder (name, is_superuser) von UM-Metadaten, liest dann die bestehenden Attribute und merged:

async def update_group(group_pk: str, **fields) -> AuthentikGroup:
    _UM_META = {"group_type", "is_admin", "discord_role_ids", "teamspeak_group_ids"}
    top_fields = {k: v for k, v in fields.items() if k not in _UM_META}
    um_fields  = {k: v for k, v in fields.items() if k in _UM_META}

    if um_fields:
        existing = await get_group_by_pk(group_pk)
        existing_attrs = {
            "um_group_type":           existing.group_type,
            "um_is_admin":             existing.is_admin,
            "um_discord_role_ids":     existing.discord_role_ids,
            "um_teamspeak_group_ids":  existing.teamspeak_group_ids,
        }
        for k, v in um_fields.items():
            existing_attrs[f"um_{k}"] = v   # z.B. "um_group_type" = "permission"
        top_fields["attributes"] = existing_attrs

    await client.patch(f"{_base()}/core/groups/{group_pk}/", json=top_fields)

Gruppenmitgliedschaften setzen

Authentik hat keinen Endpunkt zum atomaren Ersetzen aller Gruppen eines Benutzers. Das UM berechnet daher ein Diff und feuert Add/Remove parallel:

async def set_user_groups(user_pk: int, group_pks: list[str]) -> None:
    user    = await get_user_by_pk(user_pk)
    current = {str(g["pk"]) for g in user.groups_obj}
    desired = set(group_pks)
    to_add    = desired - current
    to_remove = current - desired

    # POST /core/groups/{gpk}/add_user/    {"pk": user_pk}
    # POST /core/groups/{gpk}/remove_user/ {"pk": user_pk}
    await asyncio.gather(
        *[add_user_to_group(gpk, user_pk)    for gpk in to_add],
        *[remove_user_from_group(gpk, user_pk) for gpk in to_remove],
    )

Alle UM-Attribute im Überblick

Benutzerattribute (user.attributes in Authentik):

Gruppenattribute (group.attributes in Authentik):

routers/users.py & routers/groups.py

Alle Benutzer- und Gruppendaten kommen direkt von Authentik — keine lokale Tabelle. Der Router übersetzt Pydantic-Schemas in Authentik-API-Calls.

Benutzer-Endpunkte

Besonderheit: is_active nur für Admins

if not current_user.is_admin and data.is_active is not None:
    raise HTTPException(403, "Nur Admins können den Kontostatus ändern")

Gruppen-Endpunkte

# Non-Admin: privilegierte Felder still entfernen (kein 403, nur ignorieren)
changes = data.model_dump(exclude_unset=True, exclude_none=True)
if not current_user.is_admin:
    for restricted in ("is_admin", "is_superuser"):
        changes.pop(restricted, None)

Attribute Options — Pickup-Tabellen für Dienstgrade

Datei: routers/attribute_options.py + models/attribute_option.py. Die einzige Tabelle, die lokal in PostgreSQL lebt. Sie speichert die gültigen Werte für Benutzerattribute — vergleichbar mit einem Enum, aber zur Laufzeit editierbar.

SQLAlchemy-Modell

class AttributeOption(Base):
    __tablename__ = "attribute_options"
    id            = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    attribute_key = Column(String(100))  # z.B. "um_clan_position"
    key_label     = Column(String(100))  # Anzeige: "Dienstgrad"
    value         = Column(String(100))  # intern: "gfr"
    label         = Column(String(100))  # Anzeige: "Gefreiter"
    sort_order    = Column(Integer, default=0)
    color         = Column(String(20))   # "#4e73df"
    teamspeak_group_ids = Column(JSON, default=list)
    created_at    = Column(DateTime, default=func.now())

Warum UUID als PK? Früher Integer — nach Migration auf UUID traten onclick-Handler-Bugs auf, weil JSON.stringify("uuid-string") Anführungszeichen einfügt, die das HTML-Attribut brechen. Fix: data-opt-id-Attribut + this.dataset.optId statt inline onclick-JSON.

API-Endpunkte

Verbindung zum TS3-Sync

# In users.py + teamspeak.py identisch:
async def _get_rank_ts3_groups(clan_rank: str) -> list[str]:
    async with AsyncSessionLocal() as db:
        opt = await db.execute(
            select(AttributeOption).where(
                AttributeOption.attribute_key == "um_clan_position",
                AttributeOption.value == clan_rank,  # z.B. "gfr"
            )
        )
        opt = result.scalar_one_or_none()
        return list(opt.teamspeak_group_ids) if opt else []
# → Die TS3-Gruppen des Dienstgrads fließen in desired_ts ein

services/discord.py & routers/discord.py

Zwei getrennte Auth-Flows: Bot-Token für Guild-Management (Rollen setzen), OAuth2 für das initiale Verknüpfen einer Discord-ID mit einem UM-Konto.

DiscordService — Bot-API (services/discord.py)

DISCORD_API = "https://discord.com/api/v10"

class DiscordService:
    def __init__(self, bot_token: str, guild_id: str):
        self._headers = {"Authorization": f"Bot {bot_token}"}

    # Alle Server-Rollen abrufen (ohne @everyone, sortiert nach Position)
    async def get_guild_roles(self) -> list[dict]:
        GET /guilds/{guild_id}/roles

    # Aktuelle Rollen eines Guild-Mitglieds
    async def get_member_roles(self, discord_user_id: str) -> set[str]:
        GET /guilds/{guild_id}/members/{user_id}
        → {"roles": ["role_id_1", "role_id_2", ...]}
        # 404 → leeres Set (Benutzer nicht auf Server)

    # Rollen hinzufügen / entfernen (einzeln, sequenziell — Rate-Limit-Sicherheit)
    async def add_role(self, user_id, role_id) -> bool:
        PUT /guilds/{guild_id}/members/{user_id}/roles/{role_id}
        → HTTP 204 = Erfolg

    async def remove_role(self, user_id, role_id) -> bool:
        DELETE /guilds/{guild_id}/members/{user_id}/roles/{role_id}
        → HTTP 204 = Erfolg

    async def sync_user_roles(self, user_id, old_roles: set, new_roles: set):
        for role_id in new_roles - old_roles: await self.add_role(user_id, role_id)
        for role_id in old_roles - new_roles: await self.remove_role(user_id, role_id)
        # Sequenziell, nicht parallel → vermeidet Discord Rate-Limits

# Singleton — wird beim ersten Aufruf initialisiert
def get_discord_service() -> DiscordService | None:
    if not (DISCORD_BOT_TOKEN and DISCORD_GUILD_ID): return None
    return DiscordService(DISCORD_BOT_TOKEN, DISCORD_GUILD_ID)

Discord OAuth2 Flow — ID-Verknüpfung

# GET /api/discord/oauth/url → gibt Discord-Authorize-URL zurück
state_jwt = jose_jwt.encode({
    "user_uid": current_user.id,
    "ak_pk":    current_user.ak_pk,
    "purpose":  "discord_oauth",
    "nonce":    secrets.token_hex(8),
    "exp":      utcnow() + timedelta(minutes=10),
}, SECRET_KEY, algorithm="HS256")

url = "https://discord.com/api/oauth2/authorize?" + urlencode({
    "client_id": DISCORD_CLIENT_ID,
    "redirect_uri": "{APP_BASE_URL}/api/discord/oauth/callback",
    "response_type": "code",
    "scope": "identify",   # nur User-ID lesen, kein email
    "state": state_jwt,
})

# GET /api/discord/oauth/callback?code=…&state=…
user_uid, ak_pk = _decode_state(state)  # State-JWT validieren
token_data    = await service.exchange_code(code, redirect_uri, client_id, secret)
discord_user  = await service.get_oauth_user(token_data["access_token"])
discord_id    = discord_user["id"]
await ak.update_user_attributes(ak_pk, {"um_discord_id": discord_id})

Sync-Endpunkte

services/teamspeak.py — Raw TCP ServerQuery

Kein TS3-Library-Dependency — das UM implementiert den ServerQuery-Client selbst mit asyncio.open_connection. Das ermöglicht vollständige Kontrolle ohne externe Abhängigkeiten.

Protokoll: TS3 ServerQuery

ServerQuery ist ein textbasiertes TCP-Protokoll auf Port 10011. Befehle sind Leerzeichen-getrennte key=value-Paare; mehrere Einträge werden mit | getrennt. Sonderzeichen müssen escaped werden:

Escape-Tabelle:
  \  → \\     Leerzeichen → \s
  |  → \p     \n → \n   \r → \r   \t → \t

Beispiel-Befehl:
  servergroupaddclient sgid=12 cldbid=42\n
  → Antwort: error id=0 msg=ok       (Erfolg)
  → Antwort: error id=2561 msg=...   (Fehler: client not in group)

_TS3Client — Verbindungsaufbau

async def connect(self) -> None:
    self._reader, self._writer = await asyncio.open_connection(host, port)
    await self._reader.readline()   # "TS3\n" — Begrüßungszeile (verworfen)
    await self._reader.readline()   # Welcome Banner (verworfen)

async def _exec(self, cmd: str, **kwargs) -> list[dict]:
    # Befehl senden: "cmd key=val key2=val2\n"
    self._writer.write(f"{cmd} {params}\n".encode())
    await self._writer.drain()
    # Antworten lesen bis "error id=…"
    while True:
        raw = (await self._reader.readline()).decode().strip()
        if raw.startswith("error "):
            code = int(parse(raw)["id"])
            if code != 0: raise TS3Error(code, msg)
            return data   # error id=0 = Erfolg

TeamSpeakService — sync_user_groups

async def sync_user_groups(self, ts_uid: str, desired_group_ids: list[str]) -> dict:
    async with self._connect() as client:
        # 1. TS3-UID → Datenbank-ID (cldbid) konvertieren
        #    cldbid ist eine Integer-ID in der TS3-Datenbank (≠ UID)
        cldbid = await client.client_get_dbid(ts_uid)
        #    clientgetdbidfromuid cluid={ts_uid} → {"cldbid": "42"}

        # 2. Aktuelle Servergruppen des Clients
        current = set(await client.client_servergroup_list(cldbid))
        #    servergroupsbyclientid cldbid={cldbid} → [{sgid: "12"}, ...]

        # 3. Diff berechnen
        desired = set(desired_group_ids)
        to_add    = desired - current
        to_remove = current - desired

        # 4. Entfernen (Error 2561 = "not in group" ignorieren — harmless race)
        for sgid in to_remove:
            try: await client.servergroup_del_client(sgid, cldbid)
            except TS3Error as e:
                if e.code != 2561: logger.warning(...)

        # 5. Hinzufügen
        for sgid in to_add:
            await client.servergroup_add_client(sgid, cldbid)

    return {"added": sorted(to_add), "removed": sorted(to_remove)}

Verbindungslebensdauer: Jeder Sync-Call öffnet eine neue TCP-Verbindung (asynccontextmanager _connect()) und schließt sie danach sofort mit quit\n. Kein Connection-Pool — TS3 ServerQuery-Verbindungen sind günstig und kurzlebig.

Router-Endpunkte

Desired-State-Berechnung & Auto-Trigger

Auto-Trigger (_trigger_sync in users.py)

# Wird aufgerufen wenn PUT /api/users/{pk} group_ids enthält
async def _trigger_sync(user_pk: int) -> None:
    try:
        ak_user = await ak.get_user_by_pk(user_pk)
        groups  = await ak.get_groups_for_user(user_pk)
        #   GET /api/v3/core/groups/?member_by_pk={user_pk}&page_size=200

        # Discord
        if discord_svc and ak_user.discord_id:
            desired = {r for g in groups for r in g.discord_role_ids}
            current = await discord_svc.get_member_roles(ak_user.discord_id)
            await discord_svc.sync_user_roles(ak_user.discord_id, current, desired)

        # TeamSpeak
        if ts_svc and ak_user.teamspeak_id:
            desired = {sgid for g in groups for sgid in g.teamspeak_group_ids}
            desired |= set(await _get_rank_ts3_groups(ak_user.clan_rank or ""))
            await ts_svc.sync_user_groups(ak_user.teamspeak_id, list(desired))

    except Exception as exc:
        logger.warning("Auto-sync failed pk=%s: %s", user_pk, exc)
        # Fehler werden nie nach oben propagiert — Sync ist best-effort

Desired-State-Formel

# TeamSpeak:
desired_ts = (
    {sgid for g in user_groups for sgid in g.teamspeak_group_ids}  # aus Gruppen
    | set(_get_rank_ts3_groups(user.um_clan_rank))                  # aus Dienstgrad
)

# Discord:
desired_discord = {role_id for g in user_groups for role_id in g.discord_role_ids}

Schema & Alembic

Tabelle attribute_options

attribute_options
├── id                   UUID (PK)
├── attribute_key        VARCHAR(100)  -- z.B. "um_clan_position"
├── key_label            VARCHAR(100)  -- Anzeigename der Kategorie
├── value                VARCHAR(100)  -- interner Wert (z.B. "gfr")
├── label                VARCHAR(100)  -- Anzeigename (z.B. "Gefreiter")
├── sort_order           INTEGER
├── color                VARCHAR(20)   -- Badge-Farbe (#hex)
├── teamspeak_group_ids  JSON          -- list[str]
└── created_at           TIMESTAMP

Migrationen

Alembic verwaltet alle Schema-Änderungen. Das Entrypoint-Script führt alembic upgrade head beim Containerstart automatisch aus.

Struktur & Updates

Verzeichnisstruktur

user-management/
├── backend/            # FastAPI-Anwendung (Volume-gemountet)
│   ├── app/
│   ├── alembic/
│   ├── Dockerfile
│   └── entrypoint.sh
├── frontend/           # Jinja2-Templates + Static Files (Volume-gemountet)
├── .env                # Secrets (nicht im Git!)
├── .env.example        # Vorlage (im Git)
└── docker-compose.yml

Datei-Updates (ohne Container-Restart)

Da ./backend und ./frontend als Docker-Volumes gemountet sind und Uvicorn mit --reload läuft:

• Python-Dateien (.py): Uvicorn erkennt Änderungen automatisch (~1–2 Sek.)
• HTML/JS/CSS: Sofort wirksam, kein Reload nötig
• Container-Restart erforderlich: Nur bei Änderungen an .env

Datenbankmigrationen ausführen

docker compose exec backend alembic upgrade head