integrate sub2api as upstream for auth/keys/usage via FastAPI BFF

Preserve local user table for superDream-specific features while syncing
user lifecycle, API key CRUD and usage queries through sub2api. Admin token
handles reads and user lifecycle; per-user tokens (Fernet-encrypted in DB)
handle key writes that admin endpoints do not expose.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/integrations/sub2api/__init__.py

@@ -0,0 +1,18 @@
+from app.integrations.sub2api.client import (
+    Sub2APIError,
+    Sub2APIReauthRequired,
+    Sub2APITransportError,
+    get_client,
+    close_client,
+)
+from app.integrations.sub2api import admin, user
+
+__all__ = [
+    "Sub2APIError",
+    "Sub2APIReauthRequired",
+    "Sub2APITransportError",
+    "get_client",
+    "close_client",
+    "admin",
+    "user",
+]

app/integrations/sub2api/admin.py

@@ -0,0 +1,97 @@
+"""Admin-token-authenticated calls to sub2api.
+
+Used for:
+- User lifecycle sync (create / update / delete / lookup)
+- Reading a user's API keys and usage (admin endpoints only support reads)
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from app.integrations.sub2api.client import admin_request
+
+
+# ── Users ─────────────────────────────────────────────────────────────
+
+async def create_user(
+    *,
+    email: str,
+    password: str,
+    username: str = "",
+    notes: str = "",
+    balance: float = 0,
+    concurrency: int = 0,
+    allowed_groups: list[int] | None = None,
+) -> dict[str, Any]:
+    return await admin_request(
+        "POST",
+        "/admin/users",
+        json={
+            "email": email,
+            "password": password,
+            "username": username,
+            "notes": notes,
+            "balance": balance,
+            "concurrency": concurrency,
+            "allowed_groups": allowed_groups or [],
+        },
+    )
+
+
+async def update_user(user_id: int, **fields: Any) -> dict[str, Any]:
+    """Partial update. Only non-None fields are sent."""
+    payload = {k: v for k, v in fields.items() if v is not None}
+    return await admin_request("PUT", f"/admin/users/{user_id}", json=payload)
+
+
+async def delete_user(user_id: int) -> None:
+    await admin_request("DELETE", f"/admin/users/{user_id}")
+
+
+async def get_user(user_id: int) -> dict[str, Any]:
+    return await admin_request("GET", f"/admin/users/{user_id}")
+
+
+async def find_user_by_email(email: str) -> dict[str, Any] | None:
+    data = await admin_request(
+        "GET",
+        "/admin/users",
+        params={"search": email, "page": 1, "page_size": 5},
+    )
+    items = (data or {}).get("items") or []
+    for item in items:
+        if (item.get("email") or "").lower() == email.lower():
+            return item
+    return None
+
+
+# ── API Keys (read-only from admin side) ──────────────────────────────
+
+async def list_user_api_keys(
+    user_id: int,
+    *,
+    page: int = 1,
+    page_size: int = 20,
+    sort_by: str = "created_at",
+    sort_order: str = "desc",
+) -> dict[str, Any]:
+    return await admin_request(
+        "GET",
+        f"/admin/users/{user_id}/api-keys",
+        params={
+            "page": page,
+            "page_size": page_size,
+            "sort_by": sort_by,
+            "sort_order": sort_order,
+        },
+    )
+
+
+# ── Usage (admin view per user) ───────────────────────────────────────
+
+async def get_user_usage_stats(user_id: int, period: str = "month") -> dict[str, Any]:
+    return await admin_request(
+        "GET",
+        f"/admin/users/{user_id}/usage",
+        params={"period": period},
+    )

app/integrations/sub2api/client.py

@@ -0,0 +1,131 @@
+"""Low-level HTTP client for sub2api.
+
+Responsibilities:
+- Singleton httpx.AsyncClient with configured base URL and timeout.
+- Uniform envelope parsing: {code, message, reason, metadata, data}.
+- Error translation to domain exceptions.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Mapping
+
+import httpx
+
+from app.config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+class Sub2APIError(Exception):
+    """sub2api returned a non-zero envelope code or HTTP error."""
+
+    def __init__(
+        self,
+        code: int,
+        message: str,
+        reason: str = "",
+        metadata: Mapping[str, str] | None = None,
+        http_status: int | None = None,
+    ) -> None:
+        super().__init__(f"[{code}] {message}" + (f" ({reason})" if reason else ""))
+        self.code = code
+        self.message = message
+        self.reason = reason
+        self.metadata = dict(metadata) if metadata else {}
+        self.http_status = http_status
+
+
+class Sub2APIReauthRequired(Sub2APIError):
+    """User-level token invalid/expired; caller must re-authenticate with password."""
+
+
+class Sub2APITransportError(Exception):
+    """Network / timeout / invalid JSON etc."""
+
+
+_client: httpx.AsyncClient | None = None
+
+
+def get_client() -> httpx.AsyncClient:
+    global _client
+    if _client is None or _client.is_closed:
+        _client = httpx.AsyncClient(
+            base_url=settings.sub2api_api_prefix,
+            timeout=settings.sub2api_request_timeout,
+        )
+    return _client
+
+
+async def close_client() -> None:
+    global _client
+    if _client is not None and not _client.is_closed:
+        await _client.aclose()
+    _client = None
+
+
+def _parse_envelope(resp: httpx.Response) -> Any:
+    try:
+        body = resp.json()
+    except ValueError as exc:
+        raise Sub2APITransportError(
+            f"sub2api returned non-JSON body (status={resp.status_code}): {resp.text[:200]}"
+        ) from exc
+
+    code = body.get("code", resp.status_code)
+    message = body.get("message", "")
+    reason = body.get("reason", "") or ""
+    metadata = body.get("metadata") or {}
+
+    if code == 0:
+        return body.get("data")
+
+    # Token invalidation signals from sub2api admin middleware
+    if reason in {"TOKEN_EXPIRED", "INVALID_TOKEN", "TOKEN_REVOKED", "USER_INACTIVE"}:
+        raise Sub2APIReauthRequired(code, message, reason, metadata, resp.status_code)
+
+    raise Sub2APIError(code, message, reason, metadata, resp.status_code)
+
+
+async def request(
+    method: str,
+    path: str,
+    *,
+    json: Any = None,
+    params: Mapping[str, Any] | None = None,
+    headers: Mapping[str, str] | None = None,
+) -> Any:
+    """Execute a sub2api call and return the unwrapped ``data`` field."""
+    client = get_client()
+    try:
+        resp = await client.request(
+            method,
+            path,
+            json=json,
+            params={k: v for k, v in (params or {}).items() if v is not None},
+            headers=headers,
+        )
+    except httpx.TimeoutException as exc:
+        raise Sub2APITransportError(f"sub2api timeout on {method} {path}") from exc
+    except httpx.HTTPError as exc:
+        raise Sub2APITransportError(f"sub2api transport error on {method} {path}: {exc}") from exc
+
+    return _parse_envelope(resp)
+
+
+async def admin_request(method: str, path: str, **kwargs: Any) -> Any:
+    """Issue a request authenticated with the admin API key."""
+    if not settings.sub2api_admin_token:
+        raise RuntimeError("SD_SUB2API_ADMIN_TOKEN is not configured")
+    headers = dict(kwargs.pop("headers", None) or {})
+    headers["x-api-key"] = settings.sub2api_admin_token
+    return await request(method, path, headers=headers, **kwargs)
+
+
+async def user_request(access_token: str, method: str, path: str, **kwargs: Any) -> Any:
+    """Issue a request authenticated with a user's JWT access token."""
+    if not access_token:
+        raise Sub2APIReauthRequired(401, "missing user access token", "MISSING_TOKEN")
+    headers = dict(kwargs.pop("headers", None) or {})
+    headers["Authorization"] = f"Bearer {access_token}"
+    return await request(method, path, headers=headers, **kwargs)

app/integrations/sub2api/user.py

@@ -0,0 +1,96 @@
+"""User-JWT-authenticated calls to sub2api.
+
+Used for:
+- Auth login / refresh / me (to obtain tokens for BFF proxying)
+- API Key CRUD (admin endpoints cannot create / delete keys)
+- Usage list / detail / dashboard for the authenticated user
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from app.integrations.sub2api.client import request, user_request
+
+
+# ── Auth (public; no Bearer required) ─────────────────────────────────
+
+async def login(email: str, password: str) -> dict[str, Any]:
+    """Returns AuthResponse: {access_token, refresh_token, expires_in, token_type, user}."""
+    return await request(
+        "POST",
+        "/auth/login",
+        json={"email": email, "password": password, "turnstile_token": ""},
+    )
+
+
+async def refresh_tokens(refresh_token: str) -> dict[str, Any]:
+    """Returns RefreshTokenResponse: {access_token, refresh_token, expires_in, token_type}."""
+    return await request(
+        "POST",
+        "/auth/refresh",
+        json={"refresh_token": refresh_token},
+    )
+
+
+async def logout(refresh_token: str | None = None) -> dict[str, Any]:
+    body = {"refresh_token": refresh_token} if refresh_token else {}
+    return await request("POST", "/auth/logout", json=body)
+
+
+# ── API Key CRUD ──────────────────────────────────────────────────────
+
+async def create_key(access_token: str, payload: dict[str, Any]) -> dict[str, Any]:
+    return await user_request(access_token, "POST", "/keys", json=payload)
+
+
+async def update_key(access_token: str, key_id: int, payload: dict[str, Any]) -> dict[str, Any]:
+    return await user_request(access_token, "PUT", f"/keys/{key_id}", json=payload)
+
+
+async def delete_key(access_token: str, key_id: int) -> dict[str, Any]:
+    return await user_request(access_token, "DELETE", f"/keys/{key_id}")
+
+
+async def get_key(access_token: str, key_id: int) -> dict[str, Any]:
+    return await user_request(access_token, "GET", f"/keys/{key_id}")
+
+
+# ── Groups ────────────────────────────────────────────────────────────
+
+async def list_available_groups(access_token: str) -> list[dict[str, Any]]:
+    return await user_request(access_token, "GET", "/groups/available")
+
+
+async def get_user_group_rates(access_token: str) -> dict[str, float]:
+    return await user_request(access_token, "GET", "/groups/rates")
+
+
+# ── Usage (user view) ─────────────────────────────────────────────────
+
+async def list_usage(access_token: str, **params: Any) -> dict[str, Any]:
+    return await user_request(access_token, "GET", "/usage", params=params)
+
+
+async def usage_stats(access_token: str, **params: Any) -> dict[str, Any]:
+    return await user_request(access_token, "GET", "/usage/stats", params=params)
+
+
+async def dashboard_stats(access_token: str) -> dict[str, Any]:
+    return await user_request(access_token, "GET", "/usage/dashboard/stats")
+
+
+async def dashboard_trend(access_token: str, **params: Any) -> dict[str, Any]:
+    return await user_request(access_token, "GET", "/usage/dashboard/trend", params=params)
+
+
+async def dashboard_models(access_token: str, **params: Any) -> dict[str, Any]:
+    return await user_request(access_token, "GET", "/usage/dashboard/models", params=params)
+
+
+async def dashboard_api_keys_usage(access_token: str, api_key_ids: list[int]) -> dict[str, Any]:
+    return await user_request(
+        access_token,
+        "POST",
+        "/usage/dashboard/api-keys-usage",
+        json={"api_key_ids": api_key_ids},
+    )