MZC/KnowMathBot/backend/app/deps.py

"""
FastAPI 依赖项:JWT令牌解析、当前用户加载、按角色鉴权。
"""
from __future__ import annotations
from typing import Annotated
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlmodel import Session, select
from app.core.security import decode_token
from app.db import get_session
from app.models import User, UserRole

#定义从请求头获取Token的规则
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token") #登录接口地址

#定义数据库会话和Token的依赖
SessionDep = Annotated[Session, Depends(get_session)]
TokenDep = Annotated[str, Depends(oauth2_scheme)]

def get_current_user(session: SessionDep, token: TokenDep) -> User:
    #获取当前登录用户
    try:
        payload = decode_token(token) # 解码令牌,得到用户信息
        sub = payload.get("sub") # 获取用户ID
        if not sub:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token subject")
        user_id = UUID(sub)
    except (JWTError, ValueError):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    user = session.exec(select(User).where(User.id == user_id)).first()
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

def require_role(*roles: UserRole):
    #接口角色权限控制
    def _guard(current_user: Annotated[User, Depends(get_current_user)]) -> User:
        if current_user.role not in roles:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
        return current_user
    return _guard