Building a Canonical Field-Mapping Registry

Every trading partner names the same fact differently: one retailer’s sales feed calls it ext_amt, a distributor’s EDI extract calls it TotalMonetaryAmount, and a broker’s spreadsheet calls it Net $. Wiring each of those into reconciliation with a bespoke parser produces dozens of one-off mappings that drift, contradict each other, and hide the moment a source silently renames a column mid-quarter. A canonical field-mapping registry replaces that sprawl with one version-controlled source of truth: a declarative table of entries, each stating that a specific source_field from a specific source_system becomes a specific canonical_field under a named transform at a pinned version. This page is the implementation-level companion to the field mapping strategies topic area, which frames the canonical schema this registry writes into, inside the broader data ingestion normalization pipelines discipline. It documents how to model an entry, store mappings as manifests, apply them idempotently to parsed facts, and prove coverage before any record reaches accrual.

Prerequisites

Before building the registry, confirm the following are in place:

  • Parsed, schema-tagged source records. Raw partner files must already be flattened by the CSV & EDI parsing workflows so each incoming record is a flat dict carrying source_system plus the partner’s native field names. The registry maps parsed facts, never raw X12 segments or byte offsets.
  • A ratified canonical schema. The target reconciliation schema — the set of canonical_field names every source must land in — must be agreed and frozen for the cycle. This registry maps into the schema defined by agreement schema design; it does not invent canonical fields.
  • A version-control host. Mapping manifests live as JSON in the same Git repository as the pipeline so every change is reviewed, diffed, and revertible. The registry is configuration, not runtime state.
  • Python packages: pydantic>=2.6 for the MappingEntry model and validators, and the standard-library decimal and json modules. Every monetary and quantity field flows through decimal.Decimal; never float.
  • Access role: read access to partner sample extracts and write access to the manifest directory (typically the reconciliation_etl service role via a reviewed pull request).

Step-by-Step Implementation

Step 1 — Define the canonical target schema

The registry is only meaningful against a fixed target. Declare the canonical fields, their types, and which are required for a record to be reconcilable, so coverage can later be checked mechanically rather than by inspection.

python
from decimal import Decimal

# canonical_field -> (python_type, required)
CANONICAL_SCHEMA: dict[str, tuple[type, bool]] = {
    "canonical_item":  (str,     True),
    "sold_qty":        (Decimal, True),
    "net_amount":      (Decimal, True),
    "trade_date":      (str,     True),   # ISO-8601 date
    "promo_code":      (str,     False),
    "source_system":   (str,     True),
}

REQUIRED_FIELDS = {name for name, (_, req) in CANONICAL_SCHEMA.items() if req}

Validation check: assert REQUIRED_FIELDS is non-empty and that every required field has a concrete type. A schema with no required fields cannot fail coverage, which means it cannot protect the accrual.

Step 2 — Model a mapping entry

Encode a single mapping as a Pydantic v2 model so a manifest that names an unknown canonical_field, an unregistered transform, or omits its version is rejected at load time rather than during a batch run. Each entry is a pure declaration: this source field becomes that canonical field under this transform.

python
from pydantic import BaseModel, field_validator, model_validator

TRANSFORMS = {"identity", "to_decimal", "trim_upper", "iso_date"}


class MappingEntry(BaseModel):
    source_system: str
    source_field: str
    canonical_field: str
    transform: str = "identity"
    version: int

    @field_validator("canonical_field")
    @classmethod
    def known_canonical(cls, v: str) -> str:
        if v not in CANONICAL_SCHEMA:
            raise ValueError(f"unknown canonical_field: {v}")
        return v

    @field_validator("transform")
    @classmethod
    def known_transform(cls, v: str) -> str:
        if v not in TRANSFORMS:
            raise ValueError(f"unregistered transform: {v}")
        return v

    @model_validator(mode="after")
    def version_pinned(self) -> "MappingEntry":
        if self.version < 1:
            raise ValueError("version must be a pinned positive integer")
        return self

Validation check: load an entry whose canonical_field is "net_ammount" (a typo) and assert it raises ValidationError. A mapping that points at a field the schema does not define must never reach the applier.

Step 3 — Store mappings as version-controlled manifests

Persist entries as one JSON manifest per source system, checked into Git. The file is the registry — there is no hidden database — so every rename, transform change, or override is a reviewable diff with an author and a timestamp. Load a manifest by parsing each row through MappingEntry, which enforces the Step 2 contract on the whole file.

json
[
  {"source_system": "retailer_a", "source_field": "item_upc",
   "canonical_field": "canonical_item", "transform": "trim_upper", "version": 3},
  {"source_system": "retailer_a", "source_field": "ext_amt",
   "canonical_field": "net_amount", "transform": "to_decimal", "version": 3},
  {"source_system": "retailer_a", "source_field": "qty_sold",
   "canonical_field": "sold_qty", "transform": "to_decimal", "version": 3},
  {"source_system": "retailer_a", "source_field": "sale_dt",
   "canonical_field": "trade_date", "transform": "iso_date", "version": 3}
]
python
import json
from pathlib import Path


def load_manifest(path: Path) -> list[MappingEntry]:
    rows = json.loads(path.read_text())
    entries = [MappingEntry(**row) for row in rows]
    _reject_collisions(entries)      # defined in Step 5
    return entries

Validation check: assert that two manifests for different source systems can be loaded independently and that load_manifest returns only MappingEntry instances. Per-source manifests keep one partner’s rename from mutating another partner’s mapping — the essence of a per-source override.

Step 4 — Apply mappings idempotently to parsed facts

The applier walks the entries for a record’s source_system, runs each transform, and writes the result into a fresh canonical dict. It reads from the raw input and never from its own output, so applying it twice yields the identical result — the property that lets a batch be safely replayed. Transforms that touch money or quantity return decimal.Decimal and nothing else.

python
from decimal import Decimal, InvalidOperation

def _to_decimal(raw: str) -> Decimal:
    try:
        return Decimal(str(raw))          # str() guards float inputs
    except InvalidOperation as exc:
        raise ValueError(f"non-numeric money/qty: {raw!r}") from exc

TRANSFORM_FUNCS = {
    "identity":   lambda v: v,
    "trim_upper": lambda v: str(v).strip().upper(),
    "iso_date":   lambda v: str(v).strip(),      # normalized upstream
    "to_decimal": _to_decimal,
}


def apply_mapping(raw: dict, entries: list[MappingEntry]) -> dict:
    canonical: dict = {"source_system": raw["source_system"]}
    for e in entries:
        if e.source_field in raw:
            fn = TRANSFORM_FUNCS[e.transform]
            canonical[e.canonical_field] = fn(raw[e.source_field])
    return canonical

Validation check: apply the same entries to the same record twice and assert the two output dicts are equal, and that type(out["net_amount"]) is Decimal. Idempotency plus Decimal typing is what makes a replayed batch reconcile to the penny.

Step 5 — Reject silent override collisions

A per-source override is intentional; two entries in the same manifest writing the same canonical_field is a collision that silently lets whichever ran last win. Detect it at load time so the ambiguity is a failed review, not a mystery variance three weeks later.

python
def _reject_collisions(entries: list[MappingEntry]) -> None:
    seen: dict[tuple[str, str], str] = {}
    for e in entries:
        key = (e.source_system, e.canonical_field)
        if key in seen:
            raise ValueError(
                f"collision: {e.source_system} maps both "
                f"{seen[key]!r} and {e.source_field!r} -> {e.canonical_field}"
            )
        seen[key] = e.source_field

Validation check: load a manifest where both ext_amt and net_val map to net_amount for retailer_a and assert it raises before any record is processed. A collision must fail loud at load, never resolve by row order.

Step 6 — Validate coverage against the required schema

Before a source is admitted to a run, prove its manifest fills every required canonical field. An unmapped required field is a hard stop — a record missing net_amount cannot accrue, so it must be caught at the registry boundary, not discovered downstream.

python
def validate_coverage(entries: list[MappingEntry]) -> None:
    mapped = {e.canonical_field for e in entries}
    missing = REQUIRED_FIELDS - mapped - {"source_system"}
    if missing:
        raise ValueError(f"unmapped required fields: {sorted(missing)}")


def registry_version(entries: list[MappingEntry]) -> int:
    versions = {e.version for e in entries}
    if len(versions) != 1:
        raise ValueError(f"version drift within manifest: {sorted(versions)}")
    return versions.pop()

Validation check: remove the trade_date entry from a manifest and assert validate_coverage raises unmapped required fields: ['trade_date']. Then assert registry_version raises when one entry is bumped to version 4 while its siblings stay at 3. A source with a coverage hole or mixed versions must never enter the batch.

Common Failure Modes and Fixes

  1. Unmapped required field (UNMAPPED_REQUIRED). A partner adds a required canonical field to the schema but a source manifest was never updated, so records land without net_amount and fail silently downstream. Run validate_coverage (Step 6) at load and refuse the source until every field in REQUIRED_FIELDS has an entry — a coverage gap is a blocked run, not a nullable column.
  2. Ambiguous many-to-one mapping (AMBIGUOUS_MANY_TO_ONE). Two source fields legitimately contribute to one canonical field (for example a base amount and a freight amount both feeding net_amount). Do not encode both as plain overrides that clobber each other; introduce an explicit aggregating transform whose inputs are named, so the combination is declared and reviewable rather than resolved by manifest row order.
  3. Transform losing Decimal precision (DECIMAL_PRECISION_LOSS). A to_decimal transform that routes through float(raw) first reintroduces binary rounding error that compounds across millions of lines and breaks ledger ties. Construct Decimal from the original string (Decimal(str(raw))), keep the value as Decimal through every transform, and quantize only at the final monetary boundary with ROUND_HALF_UP.
  4. Version drift between sources (VERSION_DRIFT). One partner’s manifest is bumped to a new schema version while another still emits the previous shape, so a single batch mixes incompatible field meanings. Pin one version per manifest, assert a single version per source with registry_version (Step 6), and gate the run on all admitted sources sharing a compatible registry version before merge.
  5. Silent override collision (OVERRIDE_COLLISION). Two entries in the same manifest target the same canonical_field, so the last row silently wins and the mapping becomes order-dependent. Detect the collision at load with _reject_collisions (Step 5) and require the author to delete one entry or replace both with an explicit aggregation — ambiguity should fail review, not ship. Persistent conflicts route to fallback routing logic for adjudication rather than auto-resolution.

Operational Checklist

Frequently Asked Questions

Why store mappings as version-controlled manifests instead of rows in a database? A manifest in Git makes every rename, transform swap, and override a reviewable diff with an author, a timestamp, and a revert path. A database table records the current state but not the intent or the reviewer, so a mid-quarter change to how a partner’s ext_amt is interpreted becomes an unauditable mutation. The registry is configuration, and configuration belongs under review.

How does a per-source override differ from a collision? An override is one entry in one source’s manifest choosing how that partner’s field maps, and it is isolated to that manifest by design. A collision is two entries in the same manifest both writing the same canonical field, which makes the result depend on row order. The first is intentional partner-specific behavior; the second is ambiguity, and the loader rejects it before any record is processed.

What keeps a transform from corrupting money precision? Every monetary and quantity transform constructs decimal.Decimal from the original string representation and returns Decimal, never float. Routing through float even once reintroduces binary rounding that compounds across a batch and breaks ledger ties, so the to_decimal transform guards its input with Decimal(str(raw)) and quantizes only at the final settlement boundary.

How is the registry kept in step with the canonical schema? The schema defined by agreement schema design is the single target, and coverage validation runs each source’s manifest against REQUIRED_FIELDS before admission. When the schema gains a required field, every source that fails coverage is blocked until its manifest is updated, so the registry cannot silently fall behind the reconciliation model it feeds.

Up one level: Field Mapping Strategies

Related pages

Back to