Call an async function in Pydantic custom Validator

Question

I have 2 Paydantic models Article and Author defined as follow:

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]

These models are used to parse and validate python dict that looks like:

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

to get get authors details I have an async function that from an id, it returns autho details:

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) 

to fill author detail in the article object I implemented a custom validator, so the definition of Article class:

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]

to run the code, I am using this code snippet:

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


asyncio.run(main())

this would not work as I am not awaiting get_author. when changing my code to use await get_author it will raise:

SyntaxError: asynchronous comprehension outside of an asynchronous function

which is expected as populate_author is not async function trying to await a async funtion get_author

Another alternative I tried is to make the validator async: async def populate_author(cls, value): this will raise this error:

RuntimeWarning: coroutine 'populate_author' was never awaited

which is expected as well knowing how Pydantic is implemented.

In this case, what is the solution to run an async function inside a custom Pydantic validator?

Pydatic version 1.10.14 with python 3.11

Answer 1

1. Pydantic validation must be CPU bound, not I/O bound operation. So, I think you have to modify your data as per your database outside pydantic validators. Pydantic maintainer comment related: 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.

So, I also think validation of pure data and validation as per your database, e.g, should be divided.

---

2. If you really want to implement async function call from sync function, you can use event_loop functions, such as run_until_complete, etc. Related answers here: Call async function from sync function, while the synchronous function continues : Python