I'm trying to implement async validation in FastAPI + Pydantic (v1). I need this to check uniqueness of some unique fields (username, email). I cannot just catch IntegrityError because this won't give field names which I need to return to client so frontend could highlight those fields in form.
P.S. asyncio.sleep imitates database call.
I found to possible solutions but none of them are working:
import asyncio
import concurrent.futures
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, validator
def synchronize_async_helper(to_await):
async_response = []
async def run_and_capture_result():
r = await to_await
async_response.append(r)
loop = asyncio.get_event_loop()
coroutine = run_and_capture_result()
loop.run_until_complete(coroutine)
return async_response[0]
async def validate_username_uniqueness(username):
# make database async request
async def validate_email_uniqueness(email):
# make database async request
class UserCreateSchema(BaseModel):
username: str # unique
email: EmailStr # unique
first_name: str
last_name: str
middle_name: str | None
@validator("username")
def validate_username(cls, username: str):
# method 1
# does not wait for 10 seconds
unique = synchronize_async_helper(validate_username_uniqueness())
if not unique:
raise
return username
@validator("email")
def validate_email(cls, email):
# method 2
# this blocks - one cannot open second browser tab while 10 seconds expire
pool = concurrent.futures.ThreadPoolExecutor(1)
unique = pool.submit(asyncio.run, validate_email_uniqueness(email)).result()
if not unique:
raise
return email
app = FastAPI()
@app.post('/users')
async def create_user(user_data: UserCreateSchema):
...
How to make async validation possible?
