60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""Shared FastAPI dependencies for database access and authorization."""
|
|
|
|
from collections.abc import Callable
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.exceptions import AppException
|
|
from app.core.security import decode_access_token
|
|
from app.database.session import get_db_session
|
|
from app.models.entities import UserAccount
|
|
from app.repositories.users import UserRepository
|
|
|
|
DatabaseSession = Annotated[AsyncSession, Depends(get_db_session)]
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
session: DatabaseSession,
|
|
) -> UserAccount:
|
|
"""Decode the JWT and re-check account status on every request."""
|
|
|
|
if credentials is None:
|
|
raise AppException("缺少认证令牌", code=401, status_code=401)
|
|
payload = decode_access_token(credentials.credentials)
|
|
if not payload or not payload.get("sub"):
|
|
raise AppException("无效或过期的认证令牌", code=401, status_code=401)
|
|
try:
|
|
user_id = int(payload["sub"])
|
|
except (TypeError, ValueError) as exc:
|
|
raise AppException("无效的认证令牌", code=401, status_code=401) from exc
|
|
user = await UserRepository(session).get_by_id(user_id)
|
|
if user is None or user.status != "启用":
|
|
raise AppException("用户不存在或已停用", code=401, status_code=401)
|
|
return user
|
|
|
|
|
|
def require_permission(func_key: str, *, operate: bool = False) -> Callable[..., UserAccount]:
|
|
"""Create an RBAC dependency checked against the current database state."""
|
|
|
|
async def dependency(
|
|
current_user: Annotated[UserAccount, Depends(get_current_user)],
|
|
session: DatabaseSession,
|
|
) -> UserAccount:
|
|
if current_user.role == "admin":
|
|
return current_user
|
|
statement_user = UserRepository(session)
|
|
permissions = await statement_user.permissions_for_role(current_user.role, operate=operate)
|
|
allowed = {
|
|
permission.func_key for permission in permissions
|
|
}
|
|
if func_key not in allowed:
|
|
raise AppException("无权执行该操作", code=403, status_code=403)
|
|
return current_user
|
|
|
|
return dependency
|