Schema Validation for Inventory Feeds

Every velocity score, tier boundary, and slot directive downstream is only as trustworthy as the raw feed that fed it, and warehouse feeds are hostile: legacy ERP exports drift field names between releases, WMS handhelds emit floats where the contract says integer, and a mid-night patch quietly renames a location scheme. Schema validation is the gate that refuses those payloads at the ingestion boundary instead of letting them poison a re-slotting run. This guide is part of the Velocity Data Ingestion & WMS Sync Pipelines system, and it owns the contract-enforcement layer specifically: defining the record contract, coercing and rejecting deterministically, quarantining bad records without stalling the good ones, and versioning the contract so an upstream change is caught in CI rather than in production.

What Schema Validation Means for Inventory Feeds

Schema validation for inventory feeds is the process of asserting — before a record reaches any velocity computation — that every field is present, correctly typed, within its domain range, and internally consistent, then routing the record to either the scoring path or a quarantine path based on that assertion. It is a contract, not a filter: the schema is the single authoritative statement of what a valid inventory record looks like, and both the producer (ERP/WMS extract) and the consumer (velocity engine) are held to it.

Two properties separate a production validator from an if not rec.get("sku"): continue guard scattered through the ingest code:

• Structural validation and domain validation are both mandatory. Structural checks catch the wrong shape — missing last_sync_ts, a string where an integer belongs, an unexpected extra field. Domain checks catch the wrong value inside the right shape — a negative on_hand_qty, a location_code that does not match the facility’s aisle scheme, a velocity_class outside the taxonomy. A record can be structurally perfect and still be semantically poison; both gates must fire.
• Rejection is deterministic and isolating, not fail-open. A malformed record must produce the same, explainable outcome every time and must never silently pass through with a coerced default. Equally, one bad record in a 50,000-row batch must not abort the batch — the valid rows proceed to scoring while the invalid rows are quarantined with a structured error for reconciliation.

Two enforcement positions you will meet in the field: synchronous validation at the API gateway or REST/gRPC handoff, which fails a request the instant a bad payload arrives (lowest latency to feedback, best for real-time delta feeds), and asynchronous validation inside a message-queue consumer, which validates in bulk and defers strict checks to the worker (highest throughput, best for large batch reconciliation windows). Which one you pick follows directly from the emission model you set in WMS & ERP Polling Strategies: delta-event feeds validate synchronously, snapshot batches validate in the consumer. The machinery below is the batch-consumer form; the synchronous form runs the same model per request. A separate reference on the raw contract itself lives in Validating JSON schemas for inventory updates.

Input Data Requirements

The validator consumes a list of raw inventory records as untyped dictionaries straight off the transport layer, plus the active contract version. Nothing upstream is trusted — that is the point of this layer — so every field is asserted rather than assumed. The canonical inventory record carries six load-bearing fields; the preconditions below are the quality gate the schema enforces.

Python’s pydantic is the standard tool for this workload because it fuses structural and domain validation in one declarative model, coerces where coercion is safe, and produces a structured error list that maps cleanly onto a dead-letter record. Define the contract once as a strict model and reuse it in every ingestion path so a contract change propagates from a single source.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator, ConfigDict

logger = logging.getLogger("velocity.schema")

CONTRACT_VERSION = "2.3.0"


class InventoryFeedRecord(BaseModel):
    """The authoritative contract for one inventory feed row."""
    model_config = ConfigDict(strict=True, extra="forbid")

    sku_id: str = Field(..., min_length=3, max_length=50, pattern=r"^[A-Z0-9\-]+$")
    location_code: str = Field(..., pattern=r"^[A-D]-\d{2}-\d{2}$")
    on_hand_qty: int = Field(..., ge=0)
    reserved_qty: int = Field(default=0, ge=0)
    velocity_class: str | None = Field(default=None, pattern=r"^(FAST|MED|SLOW|DEAD)$")
    last_sync_ts: datetime

    @field_validator("on_hand_qty", "reserved_qty", mode="before")
    @classmethod
    def coerce_whole_float(cls, v: object) -> object:
        """Accept 12.0 as 12, but reject 12.4 rather than truncating silently."""
        if isinstance(v, float):
            if v != int(v):
                raise ValueError("quantity must be a whole number")
            return int(v)
        return v

    @field_validator("last_sync_ts")
    @classmethod
    def require_tz(cls, v: datetime) -> datetime:
        """Normalize to UTC so downstream windowing never compares naive to aware."""
        return v.replace(tzinfo=timezone.utc) if v.tzinfo is None else v

strict=True blocks silent string-to-int coercion so a "5" from a loosely typed ERP export is a rejection, not a guess, while extra="forbid" turns an unexpected new field — the classic signature of an upstream schema change — into a loud failure rather than a silently dropped column. The two field_validator hooks handle the domain edge cases the type system alone cannot: whole-number floats and timezone normalization.

Step-by-Step Implementation

The validator runs in four passes: validate every record against the model, split the batch into a clean set and a quarantine set, ship quarantined records to a dead-letter queue with enough context to reconcile them, then stamp the clean set with the contract version before it advances. Every pass is isolated so a single bad record never aborts the batch.

1. Validate a Batch and Split Clean from Quarantined

Iterate the raw records, instantiate the model for each, and sort the outcomes into two lists. A ValidationError is caught per record — never for the batch — and unpacked into a structured, human-readable error payload keyed by the offending field so reconciliation does not require re-running the feed to see what broke.

from dataclasses import dataclass, field
from pydantic import ValidationError


@dataclass
class QuarantinedRecord:
    """A rejected row plus everything reconciliation needs to fix it."""
    record_index: int
    payload: dict
    errors: list[str] = field(default_factory=list)
    contract_version: str = CONTRACT_VERSION


def validate_batch(
    records: list[dict],
) -> tuple[list[InventoryFeedRecord], list[QuarantinedRecord]]:
    """Split a raw feed into validated records and quarantined rejects."""
    clean: list[InventoryFeedRecord] = []
    quarantined: list[QuarantinedRecord] = []
    for idx, raw in enumerate(records):
        try:
            clean.append(InventoryFeedRecord(**raw))
        except ValidationError as exc:
            errors = [f"{e['loc'][0]}: {e['msg']}" for e in exc.errors()]
            quarantined.append(QuarantinedRecord(idx, raw, errors))
    if quarantined:
        logger.warning(
            "validation quarantined %d/%d records", len(quarantined), len(records)
        )
    return clean, quarantined

2. Route Rejects to a Dead-Letter Queue

Quarantined records must land somewhere durable, queryable, and separate from transport failures so the two are diagnosed independently — a schema reject means the data is wrong, a transport failure means the pipe is wrong. The dead-letter writer records the payload, the specific field errors, and the contract version that rejected it, which is what lets a later reconciliation job tell a genuine bad row apart from a row that only failed because the contract moved.

from typing import Protocol


class DeadLetterSink(Protocol):
    def put(self, record: QuarantinedRecord) -> None: ...


def quarantine(records: list[QuarantinedRecord], sink: DeadLetterSink) -> int:
    """Persist rejects for reconciliation; return the count dead-lettered."""
    for rec in records:
        sink.put(rec)
        logger.info(
            "dead-lettered record %d (contract %s): %s",
            rec.record_index, rec.contract_version, "; ".join(rec.errors),
        )
    return len(records)

3. Deduplicate Before the Clean Set Advances

An ERP reconciliation cycle can re-emit a snapshot that was already applied. Passing duplicates downstream triggers redundant velocity recalculation and, worse, double-counted movement. Derive a composite key from sku_id, location_code, and last_sync_ts and drop rows already seen this cycle, so the scoring layer receives each state exactly once.

def dedupe(records: list[InventoryFeedRecord]) -> list[InventoryFeedRecord]:
    """Collapse re-emitted snapshots on (sku_id, location_code, last_sync_ts)."""
    seen: set[tuple[str, str, str]] = set()
    unique: list[InventoryFeedRecord] = []
    for rec in records:
        key = (rec.sku_id, rec.location_code, rec.last_sync_ts.isoformat())
        if key in seen:
            continue
        seen.add(key)
        unique.append(rec)
    dropped = len(records) - len(unique)
    if dropped:
        logger.info("deduplicated %d re-emitted records", dropped)
    return unique

4. Wire the Passes into an Ingestion Gate

The gate ties the passes together into the single entry point the consumer calls per batch: validate, quarantine the rejects, deduplicate the clean set, and hand the survivors to the scoring layer. It returns both the clean set and a rejection ratio so the caller can alert when a feed degrades — a batch that suddenly rejects 40% of its rows is an upstream contract break, not a data-quality blip.

def ingest_feed(
    records: list[dict], sink: DeadLetterSink
) -> tuple[list[InventoryFeedRecord], float]:
    """Validate, quarantine, and dedupe a raw feed; return clean rows + reject ratio."""
    clean, quarantined = validate_batch(records)
    if quarantined:
        quarantine(quarantined, sink)
    clean = dedupe(clean)
    reject_ratio = len(quarantined) / len(records) if records else 0.0
    logger.info(
        "feed ingested: %d clean, %d quarantined (%.1f%% rejected)",
        len(clean), len(quarantined), reject_ratio * 100,
    )
    return clean, reject_ratio

The clean records leave this gate and enter the scoring path described in Async Batch Processing for Velocity Recalculation, which assumes — and does not re-check — that every field type and domain constraint already holds.

Tuning & Calibration

Schema validation has fewer numeric knobs than the scoring layer, but the ones it has are consequential: how strict coercion is, what the reject-ratio alert threshold is, and whether an unknown enum value is a hard reject or a soft warning. Keep them in an externalized config so a facility with a noisier feed can loosen the alert threshold without touching code, and so the contract version is auditable.

# schema_validation.yaml — one profile per feed source
contract:
  version: "2.3.0"           # bumped on any field add/remove/retype
  extra_fields: forbid       # forbid | ignore — forbid catches upstream drift
  coerce_whole_floats: true  # accept 12.0 as 12; fractional floats always reject
quarantine:
  reject_ratio_alert: 0.05   # page when >5% of a batch is dead-lettered
  dlq_retention_days: 30     # how long rejects stay queryable for reconciliation
enums:
  velocity_class_unknown: reject   # reject | warn — warn tolerates a new tier
  location_scheme: "^[A-D]-\\d{2}-\\d{2}$"

# Equivalent Python config dict consumed by the gate
SCHEMA_VALIDATION = {
    "contract": {
        "version": "2.3.0", "extra_fields": "forbid", "coerce_whole_floats": True,
    },
    "quarantine": {"reject_ratio_alert": 0.05, "dlq_retention_days": 30},
    "enums": {
        "velocity_class_unknown": "reject",
        "location_scheme": r"^[A-D]-\d{2}-\d{2}$",
    },
}

The reject_ratio_alert is the parameter that earns its keep. A steady low-single-digit reject rate is normal background noise — a handheld mis-scan here, a truncated export there — and dead-lettering those rows is exactly right. A sudden jump past 5% almost always means the producer changed the contract without telling the consumer, and you want a page, not a silently shrinking clean set. Set the location_scheme pattern per facility: a building that uses ^[A-D]-\d{2}-\d{2}$ and one that uses zone-letter-plus-six-digit codes need different patterns, and hardcoding one guarantees the other’s valid rows get quarantined en masse.

Validation & Testing

A validator that is itself wrong is worse than none — it fails records the business needs and passes records that corrupt velocity. Assert its invariants in CI before it ever runs against a live feed. Three properties must hold: a clean record passes, each class of malformed record is quarantined with a legible error, and duplicates collapse on the composite key.

from datetime import datetime, timezone


def _valid_row() -> dict:
    return {
        "sku_id": "WIDGET-001", "location_code": "A-04-12",
        "on_hand_qty": 120, "reserved_qty": 5,
        "velocity_class": "FAST",
        "last_sync_ts": datetime(2026, 7, 2, tzinfo=timezone.utc),
    }


def test_clean_record_passes() -> None:
    clean, quarantined = validate_batch([_valid_row()])
    assert len(clean) == 1 and not quarantined
    assert clean[0].on_hand_qty == 120


def test_domain_violations_are_quarantined() -> None:
    bad_qty = _valid_row() | {"on_hand_qty": -3}
    bad_loc = _valid_row() | {"location_code": "Z-99-99"}
    frac = _valid_row() | {"on_hand_qty": 12.4}
    clean, quarantined = validate_batch([bad_qty, bad_loc, frac])
    assert not clean and len(quarantined) == 3
    assert any("on_hand_qty" in e for e in quarantined[0].errors)


def test_extra_field_is_rejected() -> None:
    # An unexpected column is the signature of an upstream contract change.
    _, quarantined = validate_batch([_valid_row() | {"warehouse_temp": 4.0}])
    assert len(quarantined) == 1


def test_duplicates_collapse_on_composite_key() -> None:
    rows = [InventoryFeedRecord(**_valid_row()), InventoryFeedRecord(**_valid_row())]
    assert len(dedupe(rows)) == 1

A healthy run: test_clean_record_passes returns one typed record, test_domain_violations_are_quarantined produces three dead-letter entries each naming its offending field, test_extra_field_is_rejected catches the phantom column, and test_duplicates_collapse_on_composite_key returns a single row. If the extra-field test starts failing after an upstream release, that is the contract drift alert doing its job — bump CONTRACT_VERSION and reconcile deliberately rather than loosening extra="forbid".

Integration Points

This layer is a boundary control, not a data source or a decision-maker. It sits between the extractors and the scoring engine, and each neighbor imposes a contract:

• Upstream extraction. The raw records and their emission cadence come from the extractors in WMS & ERP Polling Strategies. Whether you validate synchronously or in a consumer follows directly from whether that layer emits delta events or snapshot batches; validate at the same boundary where the feed lands so a bad payload never advances a watermark.
• Historical normalization. Structural validity is necessary but not sufficient for a meaningful velocity score — a correctly typed on_hand_qty still misleads if the unit of measure was never reconciled. That reconciliation is Sales History Data Mapping, which normalizes pack sizes and date formats so the fields this gate certified actually mean the same thing across sources.
• Downstream scoring. The clean set feeds Async Batch Processing for Velocity Recalculation, which explicitly assumes contract-clean input and handles only transport faults. Keep schema rejects and transport failures in separate dead-letter paths so a bad row and a dropped socket are never confused.

Further downstream, the certified velocity_class field this gate validates is the enumeration defined by the SKU Velocity Taxonomy Design layer, and the tier assignments that consume it are computed in ABC Classification Tuning for Warehouse Slotting. Validating the enum here is what keeps those two layers from inheriting an out-of-taxonomy value.

Failure Modes & Edge Cases

• Fail-open coercion masking bad data. Loose validation that coerces "5" to 5 or truncates 12.4 to 12 lets corrupt quantities flow into scoring and skew tier boundaries invisibly. Remediation: run the model with strict=True and reject fractional quantities in a field_validator rather than truncating.
• Silent column drops from an upstream schema change. A permissive schema that ignores unknown fields will quietly discard a renamed or added column, and the loss surfaces only as degraded slotting weeks later. Remediation: set extra="forbid" so a new field is a loud rejection, and treat a reject-ratio spike as a contract-drift page.
• One bad record aborting the batch. Wrapping the whole batch in a single try/except means a single malformed row discards thousands of valid ones. Remediation: catch ValidationError per record and quarantine, never per batch.
• Naive vs. aware timestamp comparison. A last_sync_ts without a timezone crashes or misorders when the windowing layer compares it against UTC-aware history. Remediation: normalize every timestamp to UTC in a validator, as require_tz does.
• Re-emitted snapshots double-counting. An ERP reconciliation cycle that resends an already-applied snapshot triggers duplicate velocity updates. Remediation: deduplicate on the sku_id + location_code + last_sync_ts composite key before the clean set advances.

FAQ

Should I use pydantic or raw JSON Schema for inventory feeds?

Use pydantic when the validated records are consumed by Python — it produces typed objects the scoring layer uses directly, fuses structural and domain checks in one model, and gives a structured error list for free. Use a raw JSON Schema document when the contract must be shared across languages or published to an external producer as the authoritative spec. Many pipelines do both: JSON Schema as the language-neutral contract of record, pydantic as the runtime enforcer inside the Python consumer. The raw-contract approach is detailed in Validating JSON schemas for inventory updates.

Where should validation run — the API gateway or the consumer?

Follow the feed shape. Real-time delta feeds validate synchronously at the gateway so a bad payload fails its request immediately and the producer sees the error at the source. Large snapshot batches validate asynchronously inside the queue consumer so one bad row never blocks the batch and bulk reconciliation stays cheap. The rule of thumb: validate at the same boundary where the feed physically lands, and never let an unvalidated record advance a sync watermark.

What do I do with records that fail validation?

Quarantine them to a dead-letter queue with the full payload, the specific field-level errors, and the contract version that rejected them — never drop them silently and never coerce them into passing. That gives reconciliation everything it needs to fix the row or the upstream producer, and it preserves an audit trail. Keep this dead-letter path separate from the one used for transport failures so a data problem and a pipe problem are diagnosed independently.

How do I evolve the schema without breaking live feeds?

Version the contract and bump the version on any field add, removal, or retype. Keep extra="forbid" so an unannounced upstream field surfaces as a loud rejection and a reject-ratio alert rather than a silent column drop. For a planned change, run the new and old contract versions in parallel over a sampled feed, confirm the reject ratio stays flat, then cut over — and stamp every dead-letter record with the version so a post-change reconciliation can tell a genuine bad row from one that only failed because the contract moved.

Why reject an unknown velocity_class instead of defaulting it?

Because a default silently reclassifies the SKU, and a mis-tiered SKU lands in the wrong zone. An unknown velocity_class almost always means the upstream taxonomy changed — a new tier was added, or a label was renamed — and that is exactly the kind of change you want a human to reconcile against the SKU Velocity Taxonomy Design layer rather than paper over. Set velocity_class_unknown: warn only in a deliberate migration window when you expect new values.

Related

• Validating JSON schemas for inventory updates — the raw JSON Schema (Draft 2020-12) contract that complements the pydantic enforcer on this page.
• WMS & ERP Polling Strategies — the extractors that emit the feed this gate validates, and whether it arrives as delta or batch.
• Sales History Data Mapping — normalizes units and dates so the validated fields mean the same thing across sources.
• Async Batch Processing for Velocity Recalculation — the scoring layer that consumes this gate’s contract-clean output.
• Velocity Data Ingestion & WMS Sync Pipelines — the parent architecture this contract-enforcement layer protects.