29 lines
1.0 KiB
Python
29 lines
1.0 KiB
Python
"""接口鉴权：从 Authorization 头取出用户中心 token，换取当前用户"""
|
|
|
|
from fastapi import Depends, Request
|
|
|
|
from app.clients.user_center import BEARER_CHALLENGE, UserCenterClient, get_user_center
|
|
from app.config import settings
|
|
from app.core.exceptions import BizException
|
|
from app.schemas.user import CurrentUser
|
|
|
|
# AUTH_ENABLED=false 时所有请求共用的本地调试用户
|
|
DEV_USER = CurrentUser(id=0, username="dev")
|
|
|
|
|
|
def _bearer_token(request: Request) -> str:
|
|
scheme, _, token = request.headers.get("Authorization", "").partition(" ")
|
|
token = token.strip()
|
|
if scheme.lower() != "bearer" or not token:
|
|
raise BizException(401, "请先登录", BEARER_CHALLENGE)
|
|
return token
|
|
|
|
|
|
async def get_current_user(
|
|
request: Request, client: UserCenterClient = Depends(get_user_center)
|
|
) -> CurrentUser:
|
|
"""同一请求内 FastAPI 会缓存结果，路由级和函数级重复声明只校验一次"""
|
|
if not settings.auth_enabled:
|
|
return DEV_USER
|
|
return await client.fetch_current_user(_bearer_token(request))
|