| 123456789101112131415161718192021222324252627282930 |
- """读取会话与写事务分开，AI 测试等慢请求不能占住数据库写锁。"""
- from fastapi import Depends
- from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
- from app.infrastructure.database import Session, transaction
- from app.infrastructure.security import decode_access, public_user
- from app.domain.models import User
- from app.shared.common import BusinessError
- async def read_db():
- async with Session() as session:
- yield session
- async def write_db():
- async with transaction() as session:
- yield session
- bearer = HTTPBearer(auto_error=False, description="登录返回 access_token，Swagger Authorize 只填令牌值")
- async def current_user(credentials: HTTPAuthorizationCredentials | None = Depends(bearer), session=Depends(read_db)) -> dict:
- """访问每次检查数据库 active 与角色，停用用户不用等 JWT 自然过期。"""
- if credentials is None or credentials.scheme.lower() != "bearer" or not credentials.credentials:
- raise BusinessError("请先登录", 401, "AUTH_REQUIRED")
- user = await session.get(User, decode_access(credentials.credentials))
- if not user or not user.active:
- raise BusinessError("账号不存在或已停用", 401, "USER_DISABLED")
- return await public_user(session, user)
|
