Versioning Compliance Metadata Schemas With Backward Compatibility #

This guide is part of the Compliance Metadata Schemas for Annual Filing area within the Core Architecture & Regulatory Mapping framework: it takes the typed data contract that gates every filing payload and answers the question that surfaces the moment a statute changes — how do you evolve that contract without invalidating the years of historical records already written under the old one?

Scope of This Page #

This page covers schema evolution: classifying a proposed change as additive or breaking, stamping every stored record with the schema version it was written under, migrating old records forward through an upcasting chain, and validating a historical payload against the version that was actually in force when it was created. It excludes the initial design of the base contract and the tiered error categorization that governs a single validation pass — both covered in the parent area — and it excludes the per-jurisdiction business rules resolved by Implementing Jurisdictional Fallback Rules for Compliance Data. Here the concern is strictly temporal: many versions of one schema coexisting across a record’s lifetime.

The Constraint That Forces Versioned Schemas #

Statutes are amended on their own cadence, and a filing record must remain interpretable against the law that governed it when it was created, not the law in force today. Delaware’s franchise-tax computation under DGCL § 503 has a statutory minimum that has been raised more than once; a 2023 filing validated against the 2023 minimum is correct, and re-validating it against a 2026 minimum would falsely flag a settled, accepted filing as non-compliant. California anchors its Statement of Information basis to the formation anniversary under Cal. Corp. Code § 1502, and the fields a payload must carry have grown as disclosure requirements expanded. The consequence for engineering is unambiguous: a stored compliance record is an immutable historical fact, and reconciliation against a prior cycle must read the rules that were actually in force at the time. That forces every record to carry the version it was written under, and it forces schema changes to be either strictly additive or accompanied by an explicit, tested migration — never a silent mutation of the shape that historical data no longer matches.

Prerequisites #

• Python 3.10+ and Pydantic v2.x — for discriminated unions, model_validator, and the strict v2 validation semantics the parent schema already relies on.
• A monotonic schema version stamped on every stored record (schema_version), never inferred at read time.
• A migration registry mapping each version to the function that upcasts a record from the prior version to it, so a v1 record can be walked forward to v3 one step at a time.
• Structured JSON logging (stdlib logging + json.dumps) so every migration is an auditable event with the from/to versions.
• An append-only store where the original record is never overwritten; upcasting produces a new in-memory view, not a destructive update of history.

Implementation: Versioned Models and an Upcasting Registry #

The module below defines three versions of one compliance schema, a registry of single-step upcasters, and a loader that reads a stored record, walks it forward to the current version, and validates it. The diagram traces a v1 record’s lineage through the migration path to a validated current model, distinguishing the additive steps from the one breaking step that needs a real transformation.

from __future__ import annotations

import json
import logging
from datetime import date
from typing import Any, Callable

from pydantic import BaseModel, Field, ValidationError, field_validator

# Structured JSON logging — each migration is an auditable from/to event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("compliance.schema_version")

CURRENT_VERSION = 3


class SchemaMigrationError(RuntimeError):
    """Raised when a stored record cannot be walked forward to the current version."""


# --- Versioned models -------------------------------------------------------
# Each version is a distinct type. The current model is the one the pipeline
# validates against; older versions exist so history stays interpretable.

class ComplianceRecordV1(BaseModel):
    schema_version: int = Field(1, frozen=True)
    entity_id: str = Field(min_length=8, max_length=12)
    jurisdiction_code: str = Field(pattern=r"^US-[A-Z]{2}$")
    filing_type: str
    fee_dollars: float                       # v1 stored the fee as a float dollar amount
    effective_date: date


class ComplianceRecordV2(BaseModel):
    schema_version: int = Field(2, frozen=True)
    entity_id: str = Field(min_length=8, max_length=12)
    jurisdiction_code: str = Field(pattern=r"^US-[A-Z]{2}$")
    filing_type: str
    fee_dollars: float
    effective_date: date
    rule_version: str                        # ADDITIVE in v2: the statute version stamp


class ComplianceRecordV3(BaseModel):
    schema_version: int = Field(3, frozen=True)
    entity_id: str = Field(min_length=8, max_length=12)
    jurisdiction_code: str = Field(pattern=r"^US-[A-Z]{2}$")
    filing_type: str
    fee_cents: int                           # BREAKING in v3: fee is now integer cents
    effective_date: date
    rule_version: str

    @field_validator("fee_cents")
    @classmethod
    def non_negative(cls, v: int) -> int:
        # Compliance-critical: a negative statutory fee is never valid.
        if v < 0:
            raise ValueError("fee_cents cannot be negative")
        return v


# --- Upcasters: each transforms a dict from version N to version N+1 ---------

def _v1_to_v2(record: dict[str, Any]) -> dict[str, Any]:
    """ADDITIVE. A new required field is defaulted; no existing field changes shape."""
    out = dict(record)
    # Records written before rule-version stamping are attributed to the earliest
    # known ruleset for their jurisdiction — a documented, auditable default.
    out.setdefault("rule_version", f"{record['jurisdiction_code'].lower()}-legacy-v0")
    out["schema_version"] = 2
    return out


def _v2_to_v3(record: dict[str, Any]) -> dict[str, Any]:
    """BREAKING. fee_dollars (float) becomes fee_cents (int); the field is renamed and
    its representation changes, so this needs a real transform, not a default."""
    out = dict(record)
    dollars = out.pop("fee_dollars")
    # Round to the nearest cent to eliminate binary-float drift before persisting.
    out["fee_cents"] = int(round(dollars * 100))
    out["schema_version"] = 3
    return out


# The registry maps a source version to its single forward step. Walking the chain
# one step at a time keeps each migration small, independently testable, and ordered.
MIGRATIONS: dict[int, Callable[[dict[str, Any]], dict[str, Any]]] = {
    1: _v1_to_v2,
    2: _v2_to_v3,
}

_MODELS: dict[int, type[BaseModel]] = {
    1: ComplianceRecordV1,
    2: ComplianceRecordV2,
    3: ComplianceRecordV3,
}


def validate_under_written_version(record: dict[str, Any]) -> BaseModel:
    """Validate a record against the version it was WRITTEN under, proving it was
    well-formed when created. Historical facts are judged by the law in force then."""
    version = record.get("schema_version")
    model = _MODELS.get(version)
    if model is None:
        raise SchemaMigrationError(f"unknown schema_version: {version!r}")
    return model.model_validate(record)


def upcast_to_current(record: dict[str, Any]) -> ComplianceRecordV3:
    """Walk a stored record forward to the current version and validate it.

    The stored record is never mutated in place — this returns a new current-shaped
    view, so the append-only history remains the source of truth.
    """
    version = record.get("schema_version")
    if not isinstance(version, int) or version not in _MODELS:
        raise SchemaMigrationError(f"unstamped or unknown schema_version: {version!r}")

    # Confirm the record was valid under its own version before migrating it forward.
    validate_under_written_version(record)

    working = dict(record)
    while working["schema_version"] < CURRENT_VERSION:
        step = working["schema_version"]
        migrate = MIGRATIONS.get(step)
        if migrate is None:
            raise SchemaMigrationError(f"no migration registered from version {step}")
        kind = "breaking" if step == 2 else "additive"
        working = migrate(working)
        logger.info(json.dumps({
            "event": "schema_upcast", "from_version": step,
            "to_version": working["schema_version"], "kind": kind,
            "entity_id": working.get("entity_id"),
        }))

    try:
        return ComplianceRecordV3.model_validate(working)
    except ValidationError as exc:
        raise SchemaMigrationError(f"upcast produced an invalid current record: {exc}") from exc


if __name__ == "__main__":
    stored_v1 = {
        "schema_version": 1,
        "entity_id": "LLC00012345",
        "jurisdiction_code": "US-DE",
        "filing_type": "franchise_tax",
        "fee_dollars": 175.00,
        "effective_date": "2023-03-01",
    }
    current = upcast_to_current(stored_v1)
    logger.info(json.dumps({
        "event": "loaded", "entity_id": current.entity_id,
        "fee_cents": current.fee_cents, "rule_version": current.rule_version,
        "schema_version": current.schema_version,
    }))

The design keeps two capabilities separate on purpose. validate_under_written_version proves a historical record was well-formed under the law that governed it — the standard reconciliation and audit need — while upcast_to_current produces a read-time view in today’s shape so live processing code only ever handles one model. Because the stored record is never overwritten, a v1 filing stays a valid v1 filing forever, and the migration is a lens over history rather than a rewrite of it.

Configuration Reference #

The rules below are the invariants that make the scheme safe; each is enforced by the module rather than left to convention.

Failure Modes and Fallback Routing #

Each fault maps onto the parent area’s three-tier categorization — CRITICAL, RECOVERABLE, AUDIT_ONLY — so a versioning failure routes the same way any other validation failure does.

1. Unstamped legacy record (CRITICAL). A record predating version stamping arrives with no schema_version. upcast_to_current raises SchemaMigrationError rather than guessing a version, because migrating a record forward from an unknown origin can silently corrupt a statutory value. The record is quarantined for a data-steward to attribute a version explicitly, exactly as the parent pipeline quarantines an invalid jurisdiction_code.
2. Missing migration step (CRITICAL). A record stamped v1 exists but only the v2→v3 upcaster is registered. The walk halts at the gap and raises rather than skipping a version, since jumping over a breaking change would apply the wrong transform. The remediation is to register the absent step and re-run; the gap is a deployment defect, not a data defect.
3. Upcast produces an invalid current record (RECOVERABLE). The chain runs but the resulting v3 payload fails current validation — a legacy fee that rounds to a value a new constraint rejects, for instance. The failure is caught and re-raised as a SchemaMigrationError carrying the underlying Pydantic errors, and the record is routed to the jurisdictional fallback resolver for bounded re-resolution before it is escalated.
4. A change misclassified as additive (AUDIT_ONLY drifting to CRITICAL). Tightening an existing field’s constraint is a breaking change even though no field is added; shipping it as “additive” lets historical records that were valid under the loose rule fail the new one. The guard is that validate_under_written_version runs before every migration, so a record that was genuinely valid when written is provable as such, and any divergence is caught in the version-parity tests rather than in production against the Entity Taxonomy & Classification inputs feeding the schema.

Frequently Asked Questions #

Related #

• Compliance Metadata Schemas for Annual Filing
• Implementing Jurisdictional Fallback Rules for Compliance Data
• Entity Taxonomy & Classification
• Core Architecture & Regulatory Mapping