95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
"""FastAPI 依赖：数据库会话包装、当前登录用户、角色校验、分页参数。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Generator
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, Header, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db as _get_db
|
|
from app.core.exceptions import AuthError, PermissionError_
|
|
from app.core.security import decode_access_token
|
|
from app.dao.account_dao import AccountDao
|
|
from app.model import Account, Role
|
|
|
|
|
|
def get_db() -> Generator[Session, None, None]:
|
|
yield from _get_db()
|
|
|
|
|
|
DbSession = Annotated[Session, Depends(get_db)]
|
|
|
|
|
|
def get_current_account(
|
|
db: DbSession,
|
|
authorization: Annotated[str | None, Header(description="Bearer <token>")] = None,
|
|
) -> Account:
|
|
"""从 Authorization 头解析登录用户。
|
|
|
|
AUTH_ENABLED=false 时直接放行（本地调试用），返回一个虚拟管理员。
|
|
"""
|
|
if not settings.AUTH_ENABLED:
|
|
return Account(id=0, username="anonymous", real_name="调试模式", role=Role.ADMIN.value)
|
|
|
|
if not authorization:
|
|
raise AuthError("请先登录（缺少 Authorization 头）")
|
|
|
|
token = authorization[7:].strip() if authorization.lower().startswith("bearer ") else authorization.strip()
|
|
payload = decode_access_token(token)
|
|
account = AccountDao.get(db, int(payload.get("sub", 0)))
|
|
if account is None or not account.is_active:
|
|
raise AuthError("账号不存在或已停用")
|
|
return account
|
|
|
|
|
|
CurrentAccount = Annotated[Account, Depends(get_current_account)]
|
|
|
|
|
|
def require_admin(account: CurrentAccount) -> Account:
|
|
if account.role != Role.ADMIN.value:
|
|
raise PermissionError_(f"该操作需要管理员权限，当前角色：{account.role}")
|
|
return account
|
|
|
|
|
|
AdminAccount = Annotated[Account, Depends(require_admin)]
|
|
|
|
|
|
def require_write(account: CurrentAccount) -> Account:
|
|
"""只读访客不允许写操作。"""
|
|
if account.role == Role.VIEWER.value:
|
|
raise PermissionError_("当前为只读账号，不能执行写操作")
|
|
return account
|
|
|
|
|
|
WriteAccount = Annotated[Account, Depends(require_write)]
|
|
|
|
|
|
def require_read(account: CurrentAccount) -> Account:
|
|
"""读操作：只要能通过登录校验就行，三种角色都可读。
|
|
|
|
这里必须单独有一个依赖，不能图省事复用 `require_write`。
|
|
早期版本把列表/详情/统计这些纯读接口全挂在了 WriteAccount 上，
|
|
结果「只读账号」登录后连一条数据都看不到 —— 权限模型里叫只读，
|
|
行为上却是「什么都不能做」，这个账号等于废的。
|
|
"""
|
|
return account
|
|
|
|
|
|
ReadAccount = Annotated[Account, Depends(require_read)]
|
|
|
|
|
|
class Pagination:
|
|
def __init__(
|
|
self,
|
|
page: Annotated[int, Query(ge=1, description="页码，从 1 开始")] = 1,
|
|
page_size: Annotated[int, Query(ge=1, le=200, description="每页条数")] = 10,
|
|
) -> None:
|
|
self.page = page
|
|
self.page_size = page_size
|
|
|
|
|
|
PageParams = Annotated[Pagination, Depends()]
|