manage/backend/src/media_library_viewer_api/clients/authentik.py

"""Authentik directory API client.

Authentik is the user-directory source (replacing the Jellyfin-backed Users
page). This client wraps the Authentik REST API for browsing the user directory
with pagination and search. OIDC authentication is unchanged — this client is
for the directory, not SSO.
"""

from __future__ import annotations

import logging
from typing import Any

import requests

logger = logging.getLogger(__name__)


class AuthentikClient:
    """Small wrapper around the Authentik core directory API."""

    def __init__(self, base_url: str, api_token: str, timeout: float = 10.0):
        if not base_url:
            raise ValueError("Authentik base_url is required")
        if not api_token:
            raise ValueError("Authentik API token is required")

        self.base_url = base_url.rstrip("/")
        if self.base_url.endswith("/api/v3"):
            self.base_url = self.base_url[:-7]
        self.api_token = api_token
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {api_token}",
                "Accept": "application/json",
            }
        )

    def get(self, path: str, **params: Any) -> Any:
        """GET an Authentik endpoint and include useful response text on errors."""
        clean_params = {k: v for k, v in params.items() if v is not None and v != ""}
        logger.debug("Authentik GET %s params=%s", path, sorted(clean_params.keys()))
        response = self.session.get(
            f"{self.base_url}/api/v3{path}",
            params=clean_params,
            timeout=self.timeout,
        )
        try:
            response.raise_for_status()
        except requests.HTTPError as exc:
            detail = response.text[:500]
            logger.warning(
                "Authentik GET %s failed status=%s url=%s",
                path,
                response.status_code,
                response.url,
            )
            raise requests.HTTPError(
                f"{response.status_code} for {response.url}: {detail}",
                response=response,
            ) from exc
        logger.debug("Authentik GET %s ok status=%s", path, response.status_code)
        return response.json()

    def users(
        self,
        search: str | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> dict[str, Any]:
        """Return a normalized page of Authentik users.

        Calls ``GET /api/v3/core/users/`` and normalizes the paginated
        Authentik response into ``{items, total, page, page_size}``. Each item
        is the raw Authentik user dict (pk, username, name, email, avatar, …)
        so the frontend can pick the fields it needs.
        """
        payload = self.get(
            "/core/users/",
            search=search,
            page=page,
            page_size=page_size,
        )
        if not isinstance(payload, dict):
            logger.warning("Authentik users payload was not a dict: %s", type(payload).__name__)
            return {"items": [], "total": 0, "page": page, "page_size": page_size}

        results = payload.get("results")
        items: list[dict[str, Any]] = (
            [item for item in results if isinstance(item, dict)] if isinstance(results, list) else []
        )

        pagination = payload.get("pagination") or {}
        total = 0
        if isinstance(pagination, dict):
            try:
                total = int(pagination.get("count") or 0)
            except (TypeError, ValueError):
                total = 0

        logger.info(
            "Authentik users page=%s page_size=%s -> %s items (total=%s)",
            page,
            page_size,
            len(items),
            total,
        )
        return {
            "items": items,
            "total": total,
            "page": page,
            "page_size": page_size,
        }