Pydantic: Is it possible to use async methods as validators?

[adpham95]

Is it possible to use async methods as validators, for instance when making a call to a DB for validating an entry exists?

• OS: Ubuntu 18.04
• Python version: 3.7.1
• Pydantic version: 0.30.0

from pydantic import BaseModel, validator

fake_db = {
    "foo": "bar"
}

async def call_to_db(id):
    return fake_db[id]

class Test(BaseModel):
    foo: str
    bar: str

    @validator('bar')
    async def bar_validator(cls, value, values):
        if values["bar"] == "foo":
            return await call_to_db("foo")

t = Test(foo="foo", bar="foo")
print(t)

Result: Test foo='foo' bar=<coroutine object Test.bar_validator at 0x7f2213ca82c8>

[samuelcolvin]

No.

That would involve the whole of pydantic being asynchronous.

You should do all asyncronous validation outside pydantic models after other validation has completely.

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.

[synchronizing]

@samuelcolvin Integrating Pydantic + Tortoise (async ORM) + FastAPI (async), how would you recommend one go about validating entries in the database? Validating database relationships/entries via Pydantic interface makes sense to me, but I'm no contributor or expert of the source.

From my understanding, the @validate_arguments decorator has asynchronous support, so why wouldn't the @validator have the same?

[AndreMPCosta]

@samuelcolvin Integrating Pydantic + Tortoise (async ORM) + FastAPI (async), how would you recommend one go about validating entries in the database? Validating database relationships/entries via Pydantic interface makes sense to me, but I'm no contributor or expert of the source.

From my understanding, the @validate_arguments decorator has asynchronous support, so why wouldn't the @validator have the same?

I have the same problem, did you find a solution @synchronizing ?

[synchronizing]

I have the same problem, did you find a solution @synchronizing ?

We are now just doing the validation within our routes.

@router.post(
    path="/",
    summary="...",
    response_model=schemas.PlatformOut,
    responses={
        status.HTTP_201_CREATED: {
            "description": "The request was successful.",
        },
        status.HTTP_409_CONFLICT: {
            "description": "A platform with this name or URL already exists.",
        },
    },
    status_code=status.HTTP_201_CREATED,
)
@version(1)
async def create_new_platform(platform: schemas.PlatformCreateIn):
    if await models.Platform.exists(name=platform.name):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A platform with this name already exists.",
        )
    elif await models.Platform.exists(url=platform.url):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A platform with this URL already exists.",
        )

    return await models.Platform.create(**platform.dict())

Not the prettiest or best solution, but it is working for us.

[AndreMPCosta]

@synchronizing I will use a similar approach, thanks!

My take:

@router.post("/register/", response_model=Player, response_model_exclude={'id'}, status_code=201)
async def player_register(player: Player):
    try:
        db_player = await Player.get_by_user(player.username)
        if db_player:
            raise UsernameNotUnique
        await player.save()
    except Error as e:
        return JSONResponse(status_code=422, content={"error": e.message(player.username)})
    return player