Amit chakraborty

Posted on Oct 14 Originally published at amitchakraborty.dev

FastAPI in production · 2. Pydantic as your API contract

#tutorial #backend #fastapiinproduction

FastAPI in production · Chapter 2 of 12 · Backend · new chapter every Wednesday night

By the end of this chapter: Use validation as the boundary between the outside world and your code.

The problem

You are a mid‑level backend engineer on a small product team. The product manager has asked you to expose a new user‑settings endpoint so the mobile app can let users pick their preferred language and notification settings. The deadline is the end of the sprint – tomorrow.

You quickly scaffold a FastAPI app, add a POST /settings route, and write the handler to accept a raw dict:

@app.post("/settings")
async def set_settings(payload: dict):
    # store payload in Redis …
    return {"status": "ok"}

When the mobile team tests the endpoint they send the following JSON:

{
  "user_id": "123",
  "language": "en",
  "notifications": {
    "email": true,
    "sms": "no"
  }
}

The request succeeds, but later you discover two problems:

The user_id is a string, not an integer, and your downstream service crashes when it tries to cast it.
The sms field is the string "no" instead of a boolean, causing a type error in the notification worker.

Because the endpoint never validated the payload, the bug only surfaces in production, leading to a hot‑fix deployment at 02:00 am, three hours of developer on‑call time, and a frustrated mobile team.

You need a way to declare the shape of the request once, have FastAPI enforce it automatically, and get clear error messages – all without writing manual checks for every field.

Before you start

Python 3.11
FastAPI 0.110.0
Pydantic 2.6.0 (installed automatically with FastAPI)
Uvicorn 0.27.0 (development server)

Verify the environment with a single command:

python -c "import fastapi, pydantic, uvicorn; print(f'FastAPI {fastapi.__version__}, Pydantic {pydantic.__version__}, Uvicorn {uvicorn.__version__}')"

Expected output (versions may differ by a patch but must be ≥ the ones listed):

FastAPI 0.110.0, Pydantic 2.6.0, Uvicorn 0.27.0

If the command fails, install the exact versions:

pip install "fastapi==0.110.0" "uvicorn[standard]==0.27.0"

Create a fresh project directory:

mkdir fastapi-pydantic-demo && cd fastapi-pydantic-demo
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt   # contains fastapi, uvicorn

Create a minimal main.py that runs:

from fastapi import FastAPI

app = FastAPI()

@app.get("/ping")
async def ping():
    return {"msg": "pong"}

Run it:

uvicorn main:app --reload

Visit http://127.0.0.1:8000/ping – you should see {"msg":"pong"}. If you do, you are ready to move on.

How do I declare the request shape with Pydantic?

What does a Pydantic model look like in FastAPI?

A Pydantic model is a regular Python class that inherits from BaseModel. FastAPI reads the type annotation of a route parameter and, if it sees a BaseModel subclass, it automatically parses, validates, and documents it.

from pydantic import BaseModel

class NotificationSettings(BaseModel):
    email: bool
    sms: bool

class SettingsRequest(BaseModel):
    user_id: int
    language: str
    notifications: NotificationSettings

When you annotate the route parameter with SettingsRequest, FastAPI will:

Parse the incoming JSON.
Validate each field according to the type hints.
Return a 422 Unprocessable Entity response if validation fails, with a JSON body that pinpoints the offending field.

Why does this matter for production?

Single source of truth – the model lives in one place, used for validation, documentation, and type‑checking in your IDE.
Predictable errors – FastAPI generates a standard error payload, which your front‑end can reliably display.
Performance – Pydantic’s compiled validators are faster than hand‑rolled if statements, especially for nested structures.

How can I see the generated OpenAPI schema?

Start the server with the model attached (see the Build it section) and visit http://127.0.0.1:8000/docs. The Swagger UI will show the request body schema derived from SettingsRequest, including the nested notifications object.

How do I customise validation behaviour?

When do I need a custom validator?

Sometimes the business rules are stricter than the basic type check. Examples:

language must be one of the ISO‑639‑1 codes (en, fr, es, …).
user_id must be a positive integer.
notifications.sms must be false for users who have not verified a phone number (which we can check against a mock function).

How do I write a validator in Pydantic 2?

Pydantic 2 uses the field_validator decorator. It receives the raw value and must either return a transformed value or raise a ValueError.

from pydantic import BaseModel, field_validator, ValidationError
from typing import Literal

class SettingsRequest(BaseModel):
    user_id: int
    language: str
    notifications: NotificationSettings

    @field_validator("user_id")
    @classmethod
    def positive_user_id(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("user_id must be a positive integer")
        return v

    @field_validator("language")
    @classmethod
    def allowed_language(cls, v: str) -> str:
        allowed = {"en", "fr", "es", "de"}
        if v not in allowed:
            raise ValueError(f"language must be one of {sorted(allowed)}")
        return v

If validation fails, the response body contains the field path and the custom message, e.g.:

{
  "detail": [
    {
      "type": "value_error",
      "loc": ["body", "language"],
      "msg": "language must be one of ['de', 'en', 'es', "fr"]",
      "input": "xx"
    }
  ]
}

How do I validate against external state?

Use a dependency that injects a service (e.g., a DB session) and call it inside the validator. Because validators are pure functions, you cannot directly access async dependencies, but you can perform a synchronous check against a cache or a pre‑loaded lookup table. For truly async validation, perform the check inside the route handler after the model has been parsed.

How do I return a response that mirrors the request schema?

Should I reuse the request model for the response?

Re‑using the request model as a response model is tempting but can expose internal fields (e.g., passwords) if the model evolves. The recommended pattern is to create a separate response model that contains only the fields you intend to expose.

class SettingsResponse(BaseModel):
    user_id: int
    language: str
    notifications: NotificationSettings
    saved_at: datetime.datetime

FastAPI will automatically convert the response object to JSON, respecting the field order and types defined in SettingsResponse.

How does FastAPI know which model to use for the response?

Add the response_model argument to the route decorator:

from datetime import datetime

@app.post("/settings", response_model=SettingsResponse)
async def set_settings(payload: SettingsRequest):
    # pretend we persisted the payload
    return SettingsResponse(
        user_id=payload.user_id,
        language=payload.language,
        notifications=payload.notifications,
        saved_at=datetime.utcnow()
    )

If the handler returns a dict that does not conform to SettingsResponse, FastAPI raises a ResponseValidationError (422) before sending anything to the client – a useful safety net in production.

Build it

Below is a complete, runnable implementation that solves the opening problem.

Create the Pydantic models – models.py

   # models.py
   from __future__ import annotations
   from datetime import datetime
   from pydantic import BaseModel, field_validator
   from typing import Literal

   class NotificationSettings(BaseModel):
       email: bool
       sms: bool

   class SettingsRequest(BaseModel):
       user_id: int
       language: str
       notifications: NotificationSettings

       @field_validator("user_id")
       @classmethod
       def positive_user_id(cls, v: int) -> int:
           if v <= 0:
               raise ValueError("user_id must be a positive integer")
           return v

       @field_validator("language")
       @classmethod
       def allowed_language(cls, v: str) -> str:
           allowed = {"en", "fr", "es", "de"}
           if v not in allowed:
               raise ValueError(f"language must be one of {sorted(allowed)}")
           return v

   class SettingsResponse(BaseModel):
       user_id: int
       language: str
       notifications: NotificationSettings
       saved_at: datetime

Wire the models into FastAPI – main.py

   # main.py
   from fastapi import FastAPI, HTTPException
   from models import SettingsRequest, SettingsResponse
   from datetime import datetime

   app = FastAPI(title="User Settings Service", version="0.1.0")

   @app.post("/settings", response_model=SettingsResponse)
   async def set_settings(payload: SettingsRequest):
       # In a real service you would persist `payload` here.
       # For the demo we just echo it back with a timestamp.
       return SettingsResponse(
           user_id=payload.user_id,
           language=payload.language,
           notifications=payload.notifications,
           saved_at=datetime.utcnow()
       )

Run the service

   uvicorn main:app --reload

Confirm validation works

Successful request:

   curl -X POST http://127.0.0.1:8000/settings \
        -H "Content-Type: application/json" \
        -d '{"user_id": 42, "language": "en", "notifications": {"email": true, "sms": false}}'

Expected output (formatted):

   {
     "user_id": 42,
     "language": "en",
     "notifications": {
       "email": true,
       "sms": false
     },
     "saved_at": "2026-10-14T12:34:56.789012Z"
   }

Invalid request (string user_id and unsupported language):

   curl -X POST http://127.0.0.1:8000/settings \
        -H "Content-Type: application/json" \
        -d '{"user_id": "abc", "language": "xx", "notifications": {"email": true, "sms": false}}'

Expected 422 response (excerpt):

   {
     "detail": [
       {
         "type": "value_error.integer",
         "loc": ["body", "user_id"],
         "msg": "value is not a valid integer",
         "input": "abc"
       },
       {
         "type": "value_error",
         "loc": ["body", "language"],
         "msg": "language must be one of ['de', 'en', 'es', 'fr']",
         "input": "xx"
       }
     ]
   }

Check the generated docs

Open http://127.0.0.1:8000/docs. The POST /settings operation shows a request body schema that matches SettingsRequest, and the response schema matches SettingsResponse. The UI also displays the custom validation messages in the Example Value section.

At this point the endpoint enforces the contract, returns clear errors, and documents itself – exactly what the mobile team needed.

When this breaks

Symptom	Cause	Fix
422 response with `"type": "value_error.jsondecode"`	The client sent malformed JSON (e.g., trailing commas).	Ensure the client sends valid JSON. In development, use `curl -v` or a proper HTTP client to see the raw payload.
422 response with `"type": "value_error.extra"`	The payload contains unexpected fields (e.g., `"extra": 1`).	By default Pydantic rejects unknown fields. If you need to allow them, set `model_config = ConfigDict(extra="allow")` in the model, but be aware this weakens contract enforcement.
`ResponseValidationError` at runtime	The route returns a dict that does not satisfy `SettingsResponse` (e.g., missing `saved_at`).	Add the missing field or change `response_model` to a model that matches the actual return shape.
Performance degradation under load	Pydantic validation is CPU‑bound; the default thread pool can become saturated when handling thousands of requests per second.	Benchmark with `locust` or `hey`. If CPU becomes a bottleneck, consider moving heavy validation (e.g., regex checks) to a background task or using `pydantic-core` compiled validators (already default in v2).
Async validator needed	Validation depends on a DB lookup (e.g., checking that `user_id` exists).	Perform the DB check inside the route after the model is parsed, not inside a Pydantic validator, because validators are synchronous. Example: `if not await db.user_exists(payload.user_id): raise HTTPException(400, "User not found")`.

What it costs

Runtime overhead – Pydantic parses and validates every request body. In a high‑throughput service (≥10 k RPS), the CPU cost can be noticeable (≈ 0.3 ms per request on a modern Intel i7). If you hit CPU limits, you may need to off‑load validation to a separate service or use a lighter schema library.
Schema rigidity – Strict validation prevents accidental contract drift, but it also makes the API less tolerant to client bugs. You must decide whether to reject unknown fields (extra="forbid" – default) or silently ignore them (extra="ignore"). The former improves data quality; the latter reduces friction for rapidly iterating front‑ends.
Maintenance burden – Separate request and response models double the number of classes you maintain. However, this separation protects you from leaking internal fields and makes versioning easier.
Error‑message size – The 422 payload can be large for deeply nested objects. In production you may want to map the error to a shorter, client‑friendly format via an exception handler.

In the interview

Typical interview prompt

“We have an endpoint that receives a JSON payload with user preferences. Show me how you would enforce the contract using FastAPI and Pydantic, and explain what happens when the client sends an invalid field.”

Weak answer

“I would create a Pydantic model with the fields and FastAPI will automatically validate it. If the client sends something wrong, FastAPI returns a 422.”

Why it’s weak: It repeats documentation verbatim, shows no understanding of the validation flow, and omits discussion of custom validators, error handling, or performance considerations.

Strong answer

“I would define a BaseModel subclass for the request, using type hints for each field. FastAPI reads the annotation and runs Pydantic’s validators before the handler executes. For business‑level rules – like restricting language codes – I’d add a @field_validator that raises a ValueError with a clear message. The framework then returns a 422 with a JSON body that pinpoints the offending field, which the front‑end can surface to the user. If we needed to check the existence of user_id in the DB, I’d perform that check after parsing because validators are synchronous. Under heavy load I’d benchmark the validation cost; if it becomes a bottleneck I could move expensive checks to a background task or use a compiled validator. Finally, I would separate request and response models to avoid leaking internal fields and to keep the contract versionable.”

How the question evolves with seniority

Junior (1–2 years) – Expected to write the model, explain the 422 response, and maybe add a simple custom validator.
Senior (3–5 years) – Must discuss trade‑offs (extra fields, performance), show how to handle async validation, and outline a strategy for versioning the contract.
Staff / Principal – Should talk about contract governance across multiple services, automated schema compatibility testing, and the impact of validation on latency budgets. They might also mention using OpenAPI code‑gen to keep client SDKs in sync.

Follow‑up that separates experience from theory

“Can you describe a time when a validation error caused a production incident, and how you mitigated it?”

A candidate who has actually shipped a service will reference a real incident (e.g., a missing bool causing a downstream crash) and explain the steps taken: adding a stricter validator, updating the client, and instrumenting a metric for validation failures.

Your tasks

Model the endpoint – Create models.py with SettingsRequest, NotificationSettings, and SettingsResponse as shown above. Add the two custom validators (user_id positive, language whitelist). Verify that uvicorn runs and that GET /docs displays the correct schema. Done when: the Swagger UI shows the nested request body and the response model.
Test validation errors – Using curl or HTTPie, send three malformed payloads:
- user_id as a string.
- language set to "zz".
- An extra field theme: "dark" at the top level. Confirm that each returns a 422 with a helpful detail entry. Done when: you can point to the exact JSON snippet for each error.
Add a synchronous DB‑existence check – Simulate a function def user_exists(uid: int) -> bool: that returns False for uid == 0. Inside the route, after the model is parsed, raise HTTPException(400, "User not found") if the check fails. Verify the 400 response. Done when: a request with "user_id": 0 returns status 400 and the body {"detail":"User not found"}.
Performance probe – Install hey (brew install hey or go install github.com/rakyll/hey@latest). Run a quick benchmark with a valid payload for 5 seconds at 100 RPS. Record the average latency. Then add a deliberately heavy validator (e.g., a regex that matches a 10 k‑character string) and re‑run the benchmark. Note the latency increase. Done when: you can quote the two latency numbers and explain the impact.
Break it – Modify SettingsResponse to omit the saved_at field, but leave the route returning the full dict. Run a request and observe the ResponseValidationError. Capture the traceback and the 422 body. Then fix the model. Done when: you have reproduced the error, captured the output, and corrected the model.

These exercises will cement the contract‑first mindset and give you concrete experience with the failure modes you’ll encounter in production.

Your tasks this week

Do the exercises above before the next chapter. Reading a tutorial and doing
one are different activities and only one of them changes what you can build.

Stuck on any of them? Say so — describe what you tried and what happened:
tell me where you got stuck. I read every one, and the questions
that come back more than twice get answered in the next chapter.