fastapi/backend/app/core/user_context.py

from fastapi import Header, Request

from app.core.config import settings
from app.core.context import UserContext
from app.core.exceptions import AppError
from app.services.external_auth_service import ExternalAuthService


async def get_user_context(
    request: Request,
    x_user_id: str | None = Header(default=None, alias="X-User-Id"),
    x_tenant_id: str | None = Header(default=None, alias="X-Tenant-Id"),
    x_user_role: str | None = Header(default=None, alias="X-User-Role"),
    x_class_id: str | None = Header(default=None, alias="X-Class-Id"),
    x_entry_scene: str | None = Header(default=None, alias="X-Entry-Scene"),
    x_request_id: str | None = Header(default=None, alias="X-Request-Id"),
) -> UserContext:
    """用户校验：正式联调优先调用 Django 用户中心，Demo 模式兼容 X-User-Id。"""
    if settings.auth_validate_enabled and (request.headers.get("Authorization") or request.headers.get("Cookie")):
        user = await ExternalAuthService().authenticate(request)
        return UserContext(
            user_id=user.user_id,
            tenant_id=user.tenant_id or x_tenant_id,
            role=user.role or x_user_role,
            class_id=x_class_id,
            entry_scene=x_entry_scene,
            request_id=x_request_id,
            ip_address=request.client.host if request.client else None,
            user_agent=request.headers.get("User-Agent"),
            username=user.username,
            display_name=user.display_name,
            auth_source=user.source,
        )

    if settings.auth_validate_enabled and not settings.auth_allow_demo_user_id:
        raise AppError("AUTH_CREDENTIAL_REQUIRED", "Authorization or Cookie is required", 401)

    if not x_user_id or not x_user_id.strip():
        raise AppError("USER_ID_REQUIRED", "X-User-Id header is required", 401)

    return UserContext(
        user_id=x_user_id.strip(),
        tenant_id=x_tenant_id,
        role=x_user_role,
        class_id=x_class_id,
        entry_scene=x_entry_scene,
        request_id=x_request_id,
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("User-Agent"),
        auth_source="demo_header",
    )