Versioning Route Payload Schemas in Python with Pydantic Migrations

This page solves one specific production failure: a schema change ships, and every dispatch record written before it becomes unparseable — so an audit that reaches back two years hits a ValidationError on the first old plan it opens.

Route and stop payloads are not static. A new field for container class, a renamed weight unit, a split of one address string into structured parts — each change would strand every record written under the old shape unless migration is designed in. The fix is to stamp every payload with a schema_version, keep a registry of small deterministic up-migration functions (v1 → v2 → v3), and chain them through a migrate_to_latest that always lands an old record on the current schema before validation. Every run emits a migration audit trail, so an auditor can see exactly which steps transformed a legacy plan. This is a core discipline of Route Schema Design within the Core Architecture & Compliance Mapping framework.

Environment & Data Prerequisites

The migration engine is built on Pydantic v2 plus the standard library. Pin the exact version so validation semantics do not drift between machines:

python -m pip install pydantic==2.9.2 pytest==8.2.0

Every persisted payload carries a schema_version integer alongside its data. The migration engine reasons over that tag and nothing else to decide which steps to run:

One rule governs every migration function: it must be pure and idempotent in effect on its version band — given a v1 body it always produces the identical v2 body, and it never mutates its input in place. A migration that reads a clock, a random source, or global state breaks the guarantee that a two-year-old record migrates the same way today as it did the day it was archived.

Step-by-Step Implementation

Step 1 — Tag payloads with a schema version

The current schema is a Pydantic model, and its version lives as a class constant so the model and its version travel together. Structured stops use a nested model; the top-level RoutePayload is what migrate_to_latest ultimately validates against.

from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from typing import Callable
from pydantic import BaseModel, Field

CURRENT_VERSION = 3


class Address(BaseModel):
    line1: str
    city: str
    postal_code: str


class Stop(BaseModel):
    stop_id: str
    address: Address
    container_class: str  # added in v2


class RoutePayload(BaseModel):
    schema_version: int = Field(default=CURRENT_VERSION)
    route_id: str
    stops: list[Stop]

Step 2 — A registry of migration functions

Each up-migration is a small function keyed by the version it upgrades from. Registering them in a dict keyed by source version — rather than a hand-written if ladder — means adding a v3→v4 step later is a one-line registration, and the chain logic never changes.

Migration = Callable[[dict], dict]
MIGRATIONS: dict[int, Migration] = {}


def migration(from_version: int) -> Callable[[Migration], Migration]:
    def register(fn: Migration) -> Migration:
        MIGRATIONS[from_version] = fn
        return fn
    return register


@migration(1)
def v1_to_v2(rec: dict) -> dict:
    out = json.loads(json.dumps(rec))  # deep copy; never mutate the input
    for stop in out["stops"]:
        stop.setdefault("container_class", "unspecified")  # backfill new field
    out["schema_version"] = 2
    return out


@migration(2)
def v2_to_v3(rec: dict) -> dict:
    out = json.loads(json.dumps(rec))
    for stop in out["stops"]:
        flat = stop.pop("address")           # was a single "line1, city ZIP" string
        line1, city_zip = [p.strip() for p in flat.split(",", 1)]
        city, postal = city_zip.rsplit(" ", 1)
        stop["address"] = {"line1": line1, "city": city.strip(), "postal_code": postal}
    out["schema_version"] = 3
    return out

Step 3 — Chain them with migrate_to_latest

The chainer walks the registry from the record’s own version up to CURRENT_VERSION, applying exactly the steps in between and recording each one. Because it looks up the next step by the record’s current version each iteration, the order is fully determined by the data — there is no ambiguity about which migrations run.

def migrate_to_latest(rec: dict) -> tuple[dict, dict]:
    """Return (migrated_record, audit) with the record on CURRENT_VERSION."""
    from_version = rec.get("schema_version", 1)
    if from_version > CURRENT_VERSION:
        raise ValueError(f"record version {from_version} is newer than {CURRENT_VERSION}")

    applied: list[str] = []
    cursor = rec
    version = from_version
    while version < CURRENT_VERSION:
        step = MIGRATIONS.get(version)
        if step is None:
            raise ValueError(f"no migration registered from version {version}")
        cursor = step(cursor)
        applied.append(step.__name__)
        version = cursor["schema_version"]

    audit = {"from_version": from_version, "to_version": version, "applied_steps": applied}
    return cursor, audit

Step 4 — Validate the migrated record

Only after migration does the record meet the current model. Validating post-migration means a legacy v1 record and a fresh v3 record follow the identical validation path, so schema drift can never let an old plan through under weaker rules.

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "logged_at": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
            "migration": getattr(record, "migration", None),
        })


audit_logger = logging.getLogger("waste_ops.schema_migrate")
if not audit_logger.handlers:
    _h = logging.StreamHandler()
    _h.setFormatter(JSONFormatter())
    audit_logger.addHandler(_h)
audit_logger.setLevel(logging.INFO)


def load_route(stored: dict) -> RoutePayload:
    migrated, audit = migrate_to_latest(stored)
    model = RoutePayload.model_validate(migrated)  # raises ValidationError if wrong
    audit_logger.info("route_migrated", extra={"migration": audit})
    return model


if __name__ == "__main__":
    legacy_v1 = {
        "schema_version": 1,
        "route_id": "R-2024-0087",
        "stops": [{"stop_id": "S1", "address": "220 Canal St, Newark 07105"}],
    }
    route = load_route(legacy_v1)
    print(route.stops[0].container_class, "|", route.stops[0].address.city)

Compliance Output

Loading the v1 record above walks it through both steps and emits one migration audit record:

{"logged_at": "2026-07-16T10:14:02.881+00:00", "level": "INFO", "event": "route_migrated", "migration": {"from_version": 1, "to_version": 3, "applied_steps": ["v1_to_v2", "v2_to_v3"]}}

Each field earns its place in the audit trail:

• from_version — the schema the record was persisted under. Retaining it proves the stored artifact was not silently rewritten; the original bytes on disk are still tagged 1, and the migration is applied at read time.
• to_version — the schema the record was interpreted under for this read. An auditor comparing two reads of the same archived plan can confirm both landed on the same to_version and therefore the same interpretation.
• applied_steps — the ordered list of migration functions that ran. This is the chain of custody for the transformation: a municipal disposal-tracking record retained for multi-year reporting must be able to show exactly how a legacy field was reinterpreted, which the JSON schema for municipal disposal tracking reference depends on when the same plan is cited years apart.

Because migration happens on read and the audit names every step, the audit report generation layer can reproduce any historical plan’s current-schema view deterministically, without ever mutating the stored record.

Verification

The tests prove the round-trip works and, critically, that each migration is idempotent in effect — running the chain on an already-current record changes nothing.

import pytest
from pydantic import ValidationError


def test_v1_migrates_and_validates():
    v1 = {
        "schema_version": 1,
        "route_id": "R-1",
        "stops": [{"stop_id": "S1", "address": "1 Dock Rd, Camden 08103"}],
    }
    route = load_route(v1)
    assert route.schema_version == 3
    assert route.stops[0].container_class == "unspecified"
    assert route.stops[0].address.postal_code == "08103"


def test_already_current_is_a_no_op():
    migrated, audit = migrate_to_latest({
        "schema_version": 3, "route_id": "R-2",
        "stops": [{"stop_id": "S1",
                   "address": {"line1": "1 Dock Rd", "city": "Camden", "postal_code": "08103"},
                   "container_class": "rolloff"}],
    })
    assert audit["applied_steps"] == []          # nothing ran
    assert migrated["schema_version"] == 3


def test_future_version_is_rejected():
    with pytest.raises(ValueError):
        migrate_to_latest({"schema_version": 99, "route_id": "R-3", "stops": []})

Common Errors

pydantic_core._pydantic_core.ValidationError: 1 validation error ... container_class Field required — a v1 record reached model_validate without being migrated first. Root cause: calling RoutePayload.model_validate(stored) directly instead of going through load_route/migrate_to_latest. Fix: never validate a stored payload directly; always run it up to the current version first, so the model only ever sees a v3-shaped body.

Migrated records drift — the same archived plan yields different output on different days. Root cause: a migration function read datetime.now() or a random default to fill a new field. Fix: migrations must be pure functions of their input; backfill new fields with a fixed sentinel like "unspecified" (Step 2) and let a later, separately-audited process enrich them, so replaying the chain is always deterministic.

Unknown fields vanish after migration. Root cause: a migration rebuilt each stop from a fixed set of keys and dropped everything it did not recognize, so a field added out-of-band was silently lost. Fix: copy the whole record and mutate only the keys the step owns (as setdefault/pop do in Step 2) rather than reconstructing it from scratch; never let a migration be the reason a field disappears without a logged reason.

FAQ

Why migrate on read instead of rewriting stored records in place?

Rewriting archived records destroys the original artifact, and for compliance the original is the evidence. Migrating on read keeps the stored bytes exactly as written — still tagged with their birth version — while presenting a current-schema view to callers. The migration audit then documents the interpretation, so you get forward compatibility without ever altering the record an auditor may later need to see unchanged.

What happens when I add a v4 schema later?

You register one @migration(3) function that turns a v3 body into v4 and bump CURRENT_VERSION to 4. The migrate_to_latest chainer needs no change — it already walks from any stored version up to whatever CURRENT_VERSION is. Existing v1, v2, and v3 records automatically gain the extra hop, and the audit trail grows the new step name.

Up: Route Schema Design

Related

• Route Schema Design — the schema discipline this versioning strategy is part of.
• JSON Schema for Municipal Disposal Tracking — the disposal-tracking record whose long retention makes versioning mandatory.
• Audit Report Generation — consumes migrated records to reproduce historical plans deterministically.
• Core Architecture & Compliance Mapping — the framework these payload schemas live inside.