X-Financial/server/src/app/api/deps.py

from collections.abc import Generator
from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, Header, HTTPException, status
from sqlalchemy.orm import Session

from app.db.session import get_session_factory


def get_db() -> Generator[Session, None, None]:
    db = get_session_factory()()
    try:
        yield db
    finally:
        db.close()


@dataclass(slots=True)
class CurrentUserContext:
    username: str
    name: str
    role_codes: list[str]
    is_admin: bool


def get_current_user(
    x_auth_username: Annotated[str | None, Header()] = None,
    x_auth_name: Annotated[str | None, Header()] = None,
    x_auth_role_codes: Annotated[str | None, Header()] = None,
    x_auth_is_admin: Annotated[str | None, Header()] = None,
) -> CurrentUserContext:
    role_codes = [item.strip() for item in (x_auth_role_codes or "").split(",") if item.strip()]
    is_admin = str(x_auth_is_admin or "").strip().lower() in {"1", "true", "yes", "on"}

    username = (x_auth_username or "").strip()
    name = (x_auth_name or username).strip()

    if not username and not name:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="请先登录后再访问知识库。",
        )

    return CurrentUserContext(
        username=username or name,
        name=name or username,
        role_codes=role_codes,
        is_admin=is_admin,
    )


def require_admin_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    if current_user.is_admin or "manager" in current_user.role_codes:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有管理员可以上传、删除或修改知识库文件。",
    )