Вызов асинхронной функции в пользовательском валидаторе Pydantic

У меня есть 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

Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
0
458
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий
  1. Проверка Pydantic должна быть привязана к процессору, а не к операции ввода-вывода. Итак, я думаю, вам нужно изменить свои данные в соответствии с вашей базой данных за пределами валидаторов 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.

Итак, я также считаю, что проверку чистых данных и проверку, например, согласно вашей базе данных, следует разделить.


  1. Если вы действительно хотите реализовать вызов асинхронной функции из функции синхронизации, вы можете использовать функции event_loop, такие как run_until_complete и т. д. Соответствующие ответы здесь: Вызов асинхронной функции из функции синхронизации, пока синхронная функция продолжается: Python

Другие вопросы по теме