FastAPI异步Pydantic验证疑问：第三方API校验卖家资质

核心问题：Pydantic不支持异步字段验证 #

Pydantic 2.x的@field_validator字段验证器是同步执行的，无法直接在其中调用异步的第三方API验证逻辑。如果强行用asyncio.run()包裹异步代码，会阻塞FastAPI的主事件循环，违背异步设计的初衷。

可行方案：分阶段异步验证 #

1. 格式校验+路由层异步业务验证 #

先用Pydantic做同步格式校验（比如TIN长度、BIC格式等），再在FastAPI路由处理函数中调用异步方法完成第三方API的校验：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, field_validator
from your_module import HTTPSessionManager

app = FastAPI()

class SellerIn(BaseModel):
    tin: str
    trrc: str
    bic: str
    bank_account: str

    # 同步格式校验
    @field_validator('tin')
    def check_tin_length(cls, v):
        if len(v) != 10:
            raise ValueError("TIN必须为10位字符")
        return v

    @field_validator('bic')
    def check_bic_length(cls, v):
        if len(v) not in (8, 11):
            raise ValueError("BIC长度需为8或11位")
        return v

# 封装异步验证逻辑
async def validate_seller_credentials(seller_data: SellerIn):
    session = HTTPSessionManager.get_session()
    
    # 验证TIN
    tin_res = await session.get(f"https://third-party-api.com/validate/tin/{seller_data.tin}")
    if tin_res.status != 200:
        raise HTTPException(status_code=400, detail="无效TIN")
    
    # 验证TRRC
    trrc_res = await session.get(f"https://third-party-api.com/validate/trrc/{seller_data.trrc}")
    if trrc_res.status != 200:
        raise HTTPException(status_code=400, detail="无效TRRC")
    
    # 验证BIC及银行资质
    bic_res = await session.get(f"https://third-party-api.com/validate/bic/{seller_data.bic}")
    if bic_res.status != 200:
        raise HTTPException(status_code=400, detail="无效BIC")

# 卖家注册路由
@app.post("/register-seller")
async def register_seller(seller_data: SellerIn):
    await validate_seller_credentials(seller_data)
    # 后续保存卖家信息等业务逻辑
    return {"msg": "卖家注册成功"}

2. 异步依赖封装验证（更优雅的路由解耦） #

用FastAPI的异步依赖封装验证逻辑，在路由中注入依赖，实现验证与业务逻辑的解耦：

from fastapi import Depends

# 异步依赖验证器
async def validate_seller_data(seller_data: SellerIn = Depends()) -> SellerIn:
    await validate_seller_credentials(seller_data)
    return seller_data

@app.post("/register-seller")
async def register_seller(seller_data: SellerIn = Depends(validate_seller_data)):
    # 验证已通过，直接处理业务
    return {"msg": "卖家注册成功"}

3. 模型层自定义异步验证方法 #

如果想把验证逻辑和模型绑定，可以在模型中定义异步验证方法，在路由中显式调用：

class SellerIn(BaseModel):
    tin: str
    trrc: str
    bic: str
    bank_account: str

    # 同步格式校验...

    async def validate_async_fields(self):
        session = HTTPSessionManager.get_session()
        if not await self._check_tin(session):
            raise ValueError("无效TIN")
        if not await self._check_trrc(session):
            raise ValueError("无效TRRC")
        if not await self._check_bic(session):
            raise ValueError("无效BIC")
    
    async def _check_tin(self, session):
        res = await session.get(f"https://third-party-api.com/validate/tin/{self.tin}")
        return res.status == 200
    
    # 同理实现_check_trrc、_check_bic方法

# 路由中调用验证
@app.post("/register-seller")
async def register_seller(seller_data: SellerIn):
    try:
        await seller_data.validate_async_fields()
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    # 业务逻辑...

避坑提醒：不要用同步方式阻塞事件循环 #

绝对不要在Pydantic验证器中用asyncio.run()执行异步代码，这会启动新的事件循环，阻塞FastAPI的主事件循环，导致高并发场景下接口响应严重延迟，完全破坏异步架构的性能优势。

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