minio-manager/app/api/deps.py

from fastapi import Depends, Header
from pydantic import BaseModel
from app.models.user import User
from app.utils.auth_utils import auth_service
from typing import Optional
import contextvars
from app.core.exceptions import ValidationException, ErrorMessage


class RequestContext(BaseModel):
    user: User
    workspace: str
    authorization: str
    source: str
    version: str

request_context_var: contextvars.ContextVar[Optional[RequestContext]] = contextvars.ContextVar(
    "request_context_var", default=None
)

async def get_request_context(
    user: User = Depends(auth_service.verify_and_get_user),
    workspace: Optional[str] = Header(None, alias="workspace"),
    authorization: Optional[str] = Header(None),
    source: Optional[str] = Header(None),
    version: Optional[str] = Header(None),
) -> RequestContext:
    """统一封装请求上下文，避免各接口重复处理头信息"""

    if not workspace:
        raise ValidationException(detail=ErrorMessage.MISSING_PARAM)
    ctx = RequestContext(
        user=user,
        workspace=workspace,
        authorization=authorization or "",
        source=source or "customer",
        version=version or "V6",
    )
    request_context_var.set(ctx)
    return ctx