FastAPI + Pydantic in Production: Contracts, Validation, and Versioning
S01E02 of FastAPI in Production
Two FastAPI services. One producing order events, one consuming them. Both running fine in staging. In production, the consumer started silently dropping records.
No exception. No 5xx. Just missing data.
The producer had shipped a field rename — total_amount to amount — in a patch release. The consumer's Pydantic model still expected total_amount. FastAPI validated the incoming payload against the consumer's model, found no total_amount, set it to None (it was Optional), and continued. No error. No alert. Downstream, every order record had a null value where revenue should be.
This is what a schema mismatch looks like in production — not a crash, but silent data corruption.
Pydantic as a Contract, Not Just a Validator
Most engineers use Pydantic to validate inputs. That's half its value. The other half is using it as an explicit contract between services — a shared definition of what data looks like, enforced at the boundary.
The mistake in the story above was Optional without a default that makes the problem visible. Here's the pattern that would have caught it:
# ❌ Silent failure — None slips through
class OrderEvent(BaseModel):
order_id: str
user_id: str
total_amount: Optional[float] = None # missing field → None, no error
# ✅ Loud failure — missing field raises immediately
class OrderEvent(BaseModel):
order_id: str
user_id: str
amount: float # required, no default — missing field → ValidationError
Required fields with no default make schema mismatches loud. If a field is genuinely optional in your domain, make it optional. If it's required, don't give it a default just to suppress validation errors.
Strict Mode for Cross-Service Boundaries
At internal service boundaries, you want Pydantic to be strict about what it accepts. Extra fields should be rejected, not silently ignored:
from pydantic import BaseModel, ConfigDict
class OrderEvent(BaseModel):
model_config = ConfigDict(extra="forbid") # reject unknown fields
order_id: str
user_id: str
amount: float
With extra="forbid", if the producer sends a field the consumer doesn't recognise, it raises immediately — forcing the teams to align rather than silently diverging. Use this on consumer models at service boundaries. Don't use it on models that accept user input, where extra fields should be ignored cleanly.
Backward-Compatible Schema Changes
Schema changes are inevitable. The question is whether they're breaking or backward-compatible. Here's the decision tree:
Adding a new optional field — backward-compatible. Existing consumers ignore it, new consumers can use it.
# v1
class OrderEvent(BaseModel):
order_id: str
amount: float
# v2 — safe, existing consumers unaffected
class OrderEvent(BaseModel):
order_id: str
amount: float
currency: str = "USD" # optional with default
Renaming a field — breaking. Handle it with a validator that accepts both during a transition window:
from pydantic import model_validator
class OrderEvent(BaseModel):
order_id: str
amount: float
@model_validator(mode="before")
@classmethod
def handle_legacy_field(cls, data):
# Accept old field name during transition
if "total_amount" in data and "amount" not in data:
data["amount"] = data.pop("total_amount")
return data
Once all producers are on the new field name, remove the validator. This gives you a controlled migration window without a hard cutover.
Removing a field — deprecate first, remove later. Mark it in the model with a comment and log a warning when it's present, so you can see when producers stop sending it before you drop it.
API Versioning Strategy
Pydantic handles schema evolution within a version. For breaking changes that can't be backward-compatible, you need versioning at the API level.
Two approaches worth knowing:
URL prefix versioning — explicit, simple, the default choice for most teams:
from fastapi import APIRouter
v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")
@v1_router.post("/orders")
async def create_order_v1(payload: OrderEventV1): ...
@v2_router.post("/orders")
async def create_order_v2(payload: OrderEventV2): ...
app.include_router(v1_router)
app.include_router(v2_router)
Run both versions simultaneously during the transition. Deprecate v1 with a response header (Deprecation: true, Sunset: <date>) so consumers know the clock is ticking.
Header-based versioning — cleaner URLs, slightly more complex routing:
from fastapi import Header, HTTPException
@app.post("/orders")
async def create_order(
payload: dict,
api_version: str = Header(default="1", alias="X-API-Version")
):
if api_version == "2":
event = OrderEventV2(**payload)
else:
event = OrderEventV1(**payload)
...
URL versioning is easier to reason about, easier to document, and easier to deprecate. Use header versioning only if you have a specific reason — cleaner client URLs, or a gateway that routes by header. Otherwise, URL prefix is the default.
Shared Models Across Services
If you control both the producer and consumer (common in a small team), consider a shared models package rather than duplicating Pydantic schemas:
pulsecart-shared/
├── pyproject.toml
└── pulsecart_shared/
└── models/
├── __init__.py
└── events.py # OrderEvent, CartAbandonedEvent, etc.
Both services install pulsecart-shared as a dependency. Schema changes happen in one place, and a version bump forces consumers to update explicitly. For a team of 4–5 engineers on a shared codebase, this is worth the overhead. For teams with independent deployment cycles, it can create coupling — weigh it against your release cadence.
The Rule
Make schema mismatches loud at the boundary, not silent in the data. Required fields stay required. Extra fields get rejected at service boundaries. Breaking changes get a transition window with a validator, not a hard cutover.
Next: S01E03 — Testing FastAPI Services: pytest setup, mocking dependencies, and integration tests in CI.
