FastAPI：能否在Pydantic验证器中访问另一个输入参数？

我是FastAPI新手，尝试将输入参数token传入Provision类并在@validator中使用，但无法在验证器中访问token，请问是否可行？

代码示例 #

from typing import Annotated 
from fastapi import Depends, FastAPI
from pydantic import BaseModel, validator

app = FastAPI()

def decryptJWT(token):
    return {"app": "app1"}

def appAuth(token):
    sub = decryptJWT(token)
    if sub["app"] in ["app1", "app2"]:
        return True
    else:
        return False

class Provision(BaseModel):
    name: str
    app: str

    @validator("app")
    def app_must_be_valid(cls, app):
        apps = appAuth(token)  # 此处无法访问token
        if app not in apps:
            raise ValueError("Invalid app name")
        return app

@app.post("/provision")
async def provision(token, provision: Provision):
    return provision

补充说明 #

我通过Authorization头获取token，代码为token: Annotated[OAuth2PasswordBearer, Depends(verifyAuthScope('write'))]。

我需要验证provision的app输入是否与用户映射的应用之一匹配，用户（clientid）与应用列表的映射存储在数据库中，无法仅通过token声明获取该信息。目前我在路径操作函数内进行验证，这里可以同时访问token和provision的app输入，但为了保持代码整洁并统一错误类型（用ValueError而非HTTPException），我希望在路径操作函数执行前提前完成输入验证。

当前路径操作函数中使用的验证代码片段（已简化错误处理和日志）：

def verify_application_mapping(provision, token):
    applications = clientid_mapped_to_application(token)
    for app in applications:
        if provision.application in app['name']:
            return True
        raise HTTPException(status_code=404, detail='Invalid Application: "{}". Valid Applications are: {}'.format(provision.application, [app['name'] for app in applications]))

    return False

直接在Pydantic模型的@validator里访问请求的token不可行，因为Pydantic模型验证独立于FastAPI请求上下文，默认拿不到请求相关的依赖参数。不过可以通过以下几种方式实现需求：

方法1：用Pydantic根验证器结合请求上下文 #

FastAPI的Request对象可以获取请求上下文，你可以在根验证器里通过context参数传递请求对象，从而拿到token：

修改模型，使用root_validator并接收values和context：

from fastapi import Request
from pydantic import root_validator

class Provision(BaseModel):
    name: str
    app: str

    @root_validator(skip_on_failure=True)
    def app_must_be_valid(cls, values, **kwargs):
        request = kwargs.get('context', {}).get('request')
        if not request:
            raise ValueError("无法获取请求上下文")
        
        # 从Authorization头解析token（假设是Bearer格式）
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            raise ValueError("无效的Authorization头")
        
        token = auth_header.split(' ')[1]
        applications = clientid_mapped_to_application(token)
        app = values.get('app')
        
        valid_apps = [app['name'] for app in applications]
        if app not in valid_apps:
            raise ValueError(f"无效的应用名称: {app}，可用应用为: {valid_apps}")
        
        return values

然后在路径操作中传递上下文：

@app.post("/provision")
async def provision(request: Request, token: Annotated[OAuth2PasswordBearer, Depends(verifyAuthScope('write'))], provision: Provision = Depends(lambda: Provision(**await request.json(), context={'request': request}))):
    return provision

这种方式稍显繁琐，更推荐下面的方法。

方法2：把验证逻辑放到FastAPI依赖中 #

将验证逻辑封装成依赖函数，同时接收token和Provision模型，依赖会在路径操作前执行，既满足提前验证的需求，又能统一错误类型：

def validate_provision_app(token: Annotated[OAuth2PasswordBearer, Depends(verifyAuthScope('write'))], provision: Provision):
    applications = clientid_mapped_to_application(token)
    valid_apps = [app['name'] for app in applications]
    if provision.app not in valid_apps:
        # 抛出ValueError，FastAPI会自动转换成422验证错误，和Pydantic原生错误格式统一
        raise ValueError(f"无效的应用名称: {provision.app}，可用应用为: {valid_apps}")
    return provision

@app.post("/provision")
async def provision(validated_provision: Provision = Depends(validate_provision_app)):
    return validated_provision

这种方式完全贴合FastAPI的依赖注入设计，代码简洁易维护，是最优方案。

方法3：自定义Pydantic字段类型 #

创建一个自定义的ValidApp字段类型，在字段的验证逻辑中获取token（需要结合请求上下文）：

from pydantic import GetCoreSchemaHandler, ValidationInfo
from pydantic_core import core_schema
from fastapi import Request

class ValidApp(str):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        return core_schema.no_info_after_validator_function(
            cls.validate,
            handler(str),
        )
    
    @classmethod
    def validate(cls, v: str, info: ValidationInfo) -> str:
        # 从上下文获取request对象
        request = info.context.get('request')
        if not request:
            raise ValueError("无法获取请求上下文")
        
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            raise ValueError("无效的Authorization头")
        
        token = auth_header.split(' ')[1]
        applications = clientid_mapped_to_application(token)
        valid_apps = [app['name'] for app in applications]
        
        if v not in valid_apps:
            raise ValueError(f"无效的应用名称: {v}，可用应用为: {valid_apps}")
        return v

class Provision(BaseModel):
    name: str
    app: ValidApp

@app.post("/provision")
async def provision(request: Request, token: Annotated[OAuth2PasswordBearer, Depends(verifyAuthScope('write'))], provision: Provision = Depends(lambda: Provision(**await request.json(), context={'request': request}))):
    return provision

这种方式适合需要复用验证逻辑的场景，但同样需要处理请求上下文传递。

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