在 Pydantic 自定义验证器中调用异步函数

Question

我有 2 个 Paydantic 模型

Article

和

Author

定义如下：

from pydantic import BaseModel
from typing import List

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


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

这些模型用于解析和验证 python 字典，如下所示：

article_data = {"id": 568, "name": "Smart People", "authors": [{"id": 123}, {"id": 456}]}

为了获取作者详细信息，我有一个异步函数，它从 id 返回作者详细信息：

async def get_author(id: int) -> Optional[dict]:
    # Simulate fetching author details from a database or other source
    authors = {
        123: {"id": 123, "first_name": "George Bob"},
        456: {"id": 456, "first_name": "Alice Smith"},
    }
    return authors.get(id) 

为了在文章对象中填充作者详细信息，我实现了一个自定义验证器，因此 Article 类的定义：

from pydantic import BaseModel, validator

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

    @validator("authors", pre=True)
    def populate_author(cls, value):
        return [Author(**(get_author(item.get("id")))) for item in value]

要运行代码，我使用以下代码片段：

async def main():
    article = Article.parse_obj(article_data)
    print(article)


asyncio.run(main())

这行不通，因为我不等待

get_author

。 当更改我的代码以使用

await get_author

时，它会引发：

SyntaxError: asynchronous comprehension outside of an asynchronous function

这是预期的，因为

populate_author

不是尝试等待异步函数的异步函数

get_author

我尝试的另一种选择是使验证器异步：

async def populate_author(cls, value):

这会引发此错误：

RuntimeWarning: coroutine 'populate_author' was never awaited

这也是我们所期望的，了解 Pydantic 是如何实现的。

在这种情况下，在自定义 Pydantic 验证器中运行异步函数的解决方案是什么？

Pydatic 版本 1.10.14 与 python 3.11

Answer 1

1. Pydantic 验证必须是 CPU 绑定的，而不是 I/O 绑定的操作。因此，我认为您必须根据数据库在 pydantic 验证器之外修改数据。 Pydantic 维护者评论相关：https://github.com/pydantic/pydantic/issues/857#issuecomment-537570546

This might sound annoying, but actually it provides a clear divide between pure data validation and checking the data is valid as per your database etc.

所以，我也认为纯数据的验证和根据数据库的验证应该分开。

---

2. 如果确实想实现同步函数调用异步函数，可以使用

   event_loop

   函数，如

   run_until_complete

   等。 相关答案在这里：从同步函数调用异步函数，而同步函数继续：Python