У меня есть 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}]}
чтобы получить сведения об авторах, у меня есть асинхронная функция, которая по идентификатору возвращает данные автора:
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)
для заполнения сведений об авторе в объекте статьи я реализовал собственный валидатор, поэтому определение класса статьи:
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






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.
Итак, я также считаю, что проверку чистых данных и проверку, например, согласно вашей базе данных, следует разделить.
event_loop, такие как run_until_complete и т. д.
Соответствующие ответы здесь: Вызов асинхронной функции из функции синхронизации, пока синхронная функция продолжается: Python