FastAPI使用OAuth2PasswordBearer如何不抛异常判断用户登录状态

你目前采用的auto_error=False的实现本身就是FastAPI官方推荐的标准做法，这个参数就是为「可选身份验证」场景专门设计的，不属于临时变通方案，可以放心在生产环境使用。

如果你的项目有不同的鉴权需求，也可以选择以下两种常见实现方式：

实现方式1：拆分两个依赖函数 #

适合绝大多数接口需要强制登录、仅少数接口需要判断登录状态的场景，将依赖拆分为强制鉴权和可选鉴权两个版本，按需调用即可：

# 强制鉴权依赖：用于必须登录才能访问的接口，验证失败直接抛401
oauth2_scheme_strict = OAuth2PasswordBearer(tokenUrl="api/users/token", auto_error=True)
def get_current_user_strict(token: str = Depends(oauth2_scheme_strict)):
    try:
        settings = get_settings()
        payload = jwt.decode(token, settings.secret_key,
                            algorithms=[settings.algorithm_hash])
        email = payload.get("email")
        if email is None:
            raise credentials_exception
        token_data = TokenData(email=email)
    except jwt.JWTError:
        raise credentials_exception
    user = UserNode.get_node_with_email(token_data.email)
    if user is None:
        raise credentials_exception
    return user

# 可选鉴权依赖：用于仅需判断登录状态的接口，验证失败返回None
oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="api/users/token", auto_error=False)
def get_current_user_optional(token: str = Depends(oauth2_scheme_optional)):
    if not token:
        return None
    try:
        settings = get_settings()
        payload = jwt.decode(token, settings.secret_key,
                            algorithms=[settings.algorithm_hash])
        email = payload.get("email")
        if email is None:
            return None
        token_data = TokenData(email=email)
    except jwt.JWTError:
        return None
    user = UserNode.get_node_with_email(token_data.email)
    return user

调用时按需选择依赖即可：

# 仅判断登录状态，不强制登录
@app.get('/')
def is_connected(user = Depends(get_current_user_optional)):
    return user

# 需要登录才能调用的接口
@app.get('/user/info')
def get_user_info(user = Depends(get_current_user_strict)):
    return user

实现方式2：自定义401异常处理器 #

仅适合临时兼容旧代码的场景，不推荐长期使用，会导致路由逻辑和异常处理逻辑耦合，维护成本高：

from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.status import HTTP_401_UNAUTHORIZED

@app.exception_handler(HTTP_401_UNAUTHORIZED)
async def custom_401_handler(request: Request, exc: HTTPException):
    # 仅针对需要判断登录状态的接口做特殊处理
    if request.url.path == "/":
        return JSONResponse(content={"login_status": False, "user": None}, status_code=200)
    # 其余接口正常返回401错误
    return JSONResponse(
        content={"detail": exc.detail},
        status_code=exc.status_code,
        headers=exc.headers
    )

选型建议 #

优先使用你当前的auto_error=False方案，或者拆分依赖的方案，这两种都是生产环境常用的标准实现，逻辑清晰无额外性能损耗。

内容的提问来源于stack exchange，提问作者rcepre