如何在Pydantic自定义Validator中调用异步函数？

Pydantic 1.x的所有校验器都是同步执行的，无法直接在校验器中调用异步函数，以下是两种可行的解决方式：

方法一：先异步填充作者数据再实例化模型 #

先批量获取所有作者的完整信息，更新原始数据后再传入模型解析：

import asyncio
from pydantic import BaseModel
from typing import List, Optional

class Author(BaseModel):
    id: int
    first_name: str

class Article(BaseModel):
    id: int
    name: str
    authors: List[Author]

async def get_author(id: int) -> Optional[dict]:
    authors = {
        123: {"id": 123, "first_name": "George Bob"},
        456: {"id": 456, "first_name": "Alice Smith"},
    }
    return authors.get(id) 

async def main():
    article_data = {"id": 568, "name": "Smart People", "authors": [{"id": 123}, {"id": 456}]}
    
    # 异步批量获取作者完整信息
    author_tasks = [get_author(auth["id"]) for auth in article_data["authors"]]
    full_authors = await asyncio.gather(*author_tasks)
    
    # 过滤获取失败的作者（可选操作）
    full_authors = [auth for auth in full_authors if auth is not None]
    
    # 更新原始数据的authors字段
    article_data["authors"] = full_authors
    
    # 实例化Article模型
    article = Article(**article_data)
    print(article.dict())

asyncio.run(main())

方法二：给Article添加异步类方法构建实例 #

在模型内部封装异步逻辑，通过自定义类方法完成作者数据填充与模型实例化：

import asyncio
from pydantic import BaseModel
from typing import List, Optional

class Author(BaseModel):
    id: int
    first_name: str

class Article(BaseModel):
    id: int
    name: str
    authors: List[Author]

    @classmethod
    async def create_with_authors(cls, data: dict):
        # 异步获取作者完整信息
        author_tasks = [get_author(auth["id"]) for auth in data["authors"]]
        full_authors = await asyncio.gather(*author_tasks)
        full_authors = [auth for auth in full_authors if auth is not None]
        
        # 更新数据并返回模型实例
        data["authors"] = full_authors
        return cls(**data)

async def get_author(id: int) -> Optional[dict]:
    authors = {
        123: {"id": 123, "first_name": "George Bob"},
        456: {"id": 456, "first_name": "Alice Smith"},
    }
    return authors.get(id) 

async def main():
    article_data = {"id": 568, "name": "Smart People", "authors": [{"id": 123}, {"id": 456}]}
    article = await Article.create_with_authors(article_data)
    print(article.dict())

asyncio.run(main())

补充说明 #

如果后续可以升级到Pydantic 2.x，它原生支持异步校验器（通过@field_validator(mode="after")装饰异步函数），但在当前1.10版本下，上述两种方式是最直接的适配方案。

内容的提问来源于stack exchange，提问作者Mehdi Ben Hamida