FastAPI中HTTPException无法中断依赖执行流问题咨询

正常情况下FastAPI会在依赖抛出HTTPException后立即终止后续依赖与路由函数的执行，但你遇到的问题可能是因为通过request.state传递数据的方式未建立明确的依赖关联。以下是几种可行的解决方案：

方案1：通过依赖注入链传递参数（推荐） #

让第二个依赖直接依赖第一个依赖的返回值，明确两者的先后执行关系，这样第一个依赖抛出异常时，第二个依赖会直接跳过执行。

修改依赖定义：

def validate_token_model(token: str = Header(...)):
    """验证Token格式，返回合法Token"""
    try:
        TokenModel(token=token)
    except RequestValidationError as error:
        logger.info(f'Token validation failed. Token:{token}. Error:{error}')
        raise HTTPException(status_code=400, detail="Incorrect token format")
    except Exception as error:
        logger.critical(f'Token validation failed. Token:{token}. Error:{error}')
        raise HTTPException(status_code=500, detail="Token validation failed.")
    return token

def get_user_by_token(token: str = Depends(validate_token_model)):
    """通过合法Token获取用户实例"""
    try:
        user = some_storage.get(token=token)
    except Exception as error:
        logger.warning(f'Failed to get user. Error:{error}')
        raise HTTPException(status_code=500, detail="Internal server error.")
    if user is None:
        logger.warning(f'User by token:{token} not exists')
        raise HTTPException(status_code=400, detail=f"User by token:{token} not exist. Login first")
    return user

路由配置只需添加最外层依赖：

@router.post("/configure_something", dependencies=[Depends(get_user_by_token)])
def configure_resource(request: Request):
    """ do something """
    return request.state.token

若仍需将Token或用户实例存入request.state，可在依赖中直接操作：

def validate_token_model(request: Request, token: str = Header(...)):
    # 验证逻辑不变
    request.state.token = token
    return token

def get_user_by_token(request: Request, token: str = Depends(validate_token_model)):
    # 获取用户逻辑不变
    request.state.user = user
    return user

方案2：合并两个依赖为单一函数 #

如果不需要拆分逻辑，可将Token验证与用户查询合并到一个依赖函数中，确保异常抛出后直接终止后续执行：

def validate_token_and_get_user(request: Request, token: str = Header(...)):
    # 1. 验证Token格式
    try:
        TokenModel(token=token)
    except RequestValidationError as error:
        logger.info(f'Token validation failed. Token:{token}. Error:{error}')
        raise HTTPException(status_code=400, detail="Incorrect token format")
    except Exception as error:
        logger.critical(f'Token validation failed. Token:{token}. Error:{error}')
        raise HTTPException(status_code=500, detail="Token validation failed.")
    request.state.token = token

    # 2. 通过Token获取用户
    try:
        user = some_storage.get(token=token)
    except Exception as error:
        logger.warning(f'Failed to get user. Error:{error}')
        raise HTTPException(status_code=500, detail="Internal server error.")
    if user is None:
        logger.warning(f'User by token:{token} not exists')
        raise HTTPException(status_code=400, detail=f"User by token:{token} not exist. Login first")
    request.state.user = user

路由配置简化为：

@router.post("/configure_something", dependencies=[Depends(validate_token_and_get_user)])
def configure_resource(request: Request):
    """ do something """
    return request.state.token

方案3：检查并更新FastAPI版本 #

若以上方法无效，可能是使用的FastAPI版本存在依赖执行逻辑bug，请升级至最新稳定版本：

pip install --upgrade fastapi uvicorn

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