# FastAPI + Pydantic in Production: Contracts, Validation, and Versioning

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:

```python
# ❌ 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:

```python
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.

```python
# 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:

```python
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:

```python
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:

```python
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:

```plaintext
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.**
