Pydantic v2 Async Custom Validator: What to Do Instead

Key takeaways:

  • Pydantic v2 validators are synchronous — they cannot be async or await.
  • Keep field_validator and model_validator for pure shape checks.
  • Run async rules such as uniqueness in the service layer or a dependency.
  • Express request-time async preconditions as FastAPI dependencies.
  • Test the synchronous model rules and the async service rules separately.

This guide corrects a common expectation around Custom Validators and Field Constraints and routes async checks to where they belong.

The Problem This Solves

Developers reach for an async def field_validator to check uniqueness or call an external service during validation, and it silently does not work — Pydantic does not await it. The real need is legitimate; it just belongs outside the model. This guide shows the correct placement.

Prerequisites

  • Pydantic v2 and an async FastAPI stack.
  • An async database session available via dependency injection.

Step-by-Step Implementation

1. Keep the model synchronous

from pydantic import BaseModel, EmailStr, field_validator


class SignupRequest(BaseModel):
    email: EmailStr

    @field_validator("email")
    @classmethod
    def normalize(cls, v: str) -> str:
        return v.lower()          # Pure, synchronous shape normalization only.

2. Put the async rule in the service

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession


async def register(session: AsyncSession, data: SignupRequest) -> int:
    # The async uniqueness check belongs here, with a real session.
    if await email_exists(session, data.email):
        raise HTTPException(409, "email already registered")
    return await insert_user(session, data.email)

3. Or express it as a dependency

from typing import Annotated

from fastapi import Depends


async def unique_email(data: SignupRequest, session: SessionDep) -> SignupRequest:
    # Runs before the handler; raises a structured error on conflict.
    if await email_exists(session, data.email):
        raise HTTPException(409, "email already registered")
    return data


@router.post("/signup")
async def signup(data: Annotated[SignupRequest, Depends(unique_email)],
                 session: SessionDep) -> dict[str, int]:
    return {"id": await register(session, data)}

Edge Cases and Gotchas

  • Silent no-op. An async def validator is not awaited by Pydantic; do not rely on it.
  • Validation vs authorization. Async checks that depend on the caller's identity are authorization, not validation; keep them in dependencies.
  • Error consistency. Raise a domain error so the response uses your error envelope.

Verification

import pytest
from fastapi import HTTPException


async def test_duplicate_email_rejected(session):
    await insert_user(session, "ada@example.com")
    with pytest.raises(HTTPException):
        await register(session, SignupRequest(email="ADA@example.com"))