ajyy/backend/app/core/deps.py

"""FastAPI 依赖：当前用户与菜单权限校验"""
from typing import Annotated, Callable

from fastapi import Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.response import AppHTTPError
from app.core.security import TokenError, safe_decode_token
from app.crud import menu as menu_crud
from app.crud import user as user_crud
from app.models.system import SysUser

bearer_scheme = HTTPBearer(auto_error=False)


def get_current_user(
    db: Annotated[Session, Depends(get_db)],
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
) -> SysUser:
    if creds is None or not creds.credentials:
        raise AppHTTPError(401, "未登录或令牌无效")
    try:
        payload = safe_decode_token(creds.credentials)
    except TokenError:
        raise AppHTTPError(401, "未登录或令牌无效") from None
    user_id = payload.get("sub")
    if not user_id:
        raise AppHTTPError(401, "未登录或令牌无效")
    user = user_crud.get_user_by_id(db, user_id)
    if not user or user.status != "0":
        raise AppHTTPError(401, "未登录或令牌无效")
    return user


def require_menu(*codes: str) -> Callable:
    """当前用户角色若不拥有 codes 中任一权限点 → 403。"""
    needed = set(codes)

    def _dep(
        user: Annotated[SysUser, Depends(get_current_user)],
        db: Annotated[Session, Depends(get_db)],
    ) -> SysUser:
        if not menu_crud.role_has_any_menu_code(db, user.role_id, needed):
            raise AppHTTPError(403, "无权限")
        return user

    return _dep