dekun
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
"""从 Authorization 头解析当前用户，并区分管理员。"""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.errors import AppError
|
|
from app.models import User, UserRole
|
|
from app.security import read_user_id
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_current_user(
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> User:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise AppError("未登录", 401)
|
|
user = db.get(User, read_user_id(credentials.credentials))
|
|
if user is None:
|
|
raise AppError("登录已失效", 401)
|
|
return user
|
|
|
|
|
|
def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
|
|
if user.role != UserRole.ADMIN:
|
|
raise AppError("需要管理员权限", 403)
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_user)]
|
|
AdminUser = Annotated[User, Depends(require_admin)]
|