Schema Validation for Ingested Records

Operating within the broader Catalog Ingestion & ILS Sync Pipelines architecture, schema validation is the gate that decides which records are allowed to become catalog state and which are turned away at the door. It sits immediately after Parsing MARC Records with pymarc has produced structured records and before any delta reaches the ILS: parsing answers “is this a well-formed MARC record?”, while validation answers the harder question “does this record satisfy the contract my catalog, my indexer, and my compliance policy require?”. Library-tech staff hit this problem the first time a vendor feed silently ships records with a control field carrying subfields, a 650 heading pointing at an unauthorized thesaurus, or a patron identifier smuggled into a local note — each of which parses cleanly and then corrupts the catalog days later. This page walks through defining the validation contract, building the tiered validator, masking PII, routing failures to quarantine, and proving the whole thing works, with production-grade Python throughout.

Specification & Contract

A schema validator for MARC ingestion enforces three separable contracts, and conflating them is the most common design mistake. The syntactic contract governs bytes and encoding: valid UTF-8, leader length exactly 24, directory arithmetic that resolves. The structural contract governs shape: which tags are mandatory, whether a field is a control field or a data field, indicator and subfield counts. The semantic contract governs meaning: whether a subject heading resolves to an authorized vocabulary, whether a call number matches local shelving rules, whether a linked authority ID exists. Each tier has a different failure cost and a different remediation path, so the validator must report which tier rejected a record, not just that it failed.

The table below is the minimum field-level contract the validator enforces before a bibliographic record is accepted. Full tag-to-attribute translation is owned by MARC21 field mapping for modern pipelines; this contract only asserts presence, cardinality, and structural class.

Control fields (001–009) carry a single data string; data fields (010–999) carry two indicators and one or more subfields. A record that presents subfields on a control tag is structurally impossible and must be rejected outright rather than coerced, because coercion hides a corrupt upstream export. The byte-level leader contract summarized in the first row is applied in full by the downstream gate, Validating MARC Leader Fields Before Database Insert, which this page’s structural tier delegates to.

Prerequisites & Environment Setup

The examples target Python 3.11+ and Pydantic v2, whose validation model and model_serializer API differ substantially from Pydantic v1 — code written against v1 @validator decorators will not run here, so pin the major version. Authority lookups and the quarantine sink are injected through environment variables so the same worker image runs unchanged across staging and production, and the PII hashing salt is supplied as a secret, never committed.

• Python >= 3.11 available in the worker image (python --version)
• pydantic >= 2.5 pinned in requirements.txt (v2 field_validator / model_serializer APIs are assumed throughout)
• structlog >= 24.1 installed for JSON-formatted, SIEM-queryable audit logs
• SCHEMA_QUARANTINE_URL set to the broker/queue endpoint that receives rejected payloads
• AUTHORITY_STORE_DSN pointing at the controlled-vocabulary lookup service or cache
• PII_HASH_SALT exported as a secret (32+ random bytes, never committed)
• INGEST_CORRELATION_ID propagated from the batch orchestrator for audit tracing
• Validator runs with no write access to the ILS — it emits accepted deltas onto the broker, it never inserts directly

python -m venv .venv && source .venv/bin/activate
pip install "pydantic>=2.5,<3" "structlog>=24.1"
export SCHEMA_QUARANTINE_URL="amqp://ingest:***@broker.internal:5672/quarantine"
export AUTHORITY_STORE_DSN="postgresql://[email protected]:5432/vocab"
export PII_HASH_SALT="$(openssl rand -hex 32)"

Core Implementation

Step 1 — Declare the contract as typed models

Validation begins with declarative contract definitions. Model MARC21 records as Pydantic classes with explicit field-level constraints, type coercion rules, and conditional logic. extra="forbid" blocks unexpected vendor extensions from silently entering the pipeline, and a field validator isolates control fields from data fields at the model boundary.

from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator


class MarcSubfield(BaseModel):
    code: str = Field(pattern=r"^[a-z0-9]$")
    value: str = Field(min_length=1, max_length=2048)


class MarcField(BaseModel):
    tag: str = Field(pattern=r"^\d{3}$")
    indicator1: Optional[str] = Field(default=None, pattern=r"^[ #0-9a-z]$")
    indicator2: Optional[str] = Field(default=None, pattern=r"^[ #0-9a-z]$")
    subfields: list[MarcSubfield] = Field(default_factory=list)

    @field_validator("tag")
    @classmethod
    def reject_control_field_with_subfields(cls, v: str) -> str:
        # Control fields (00X) carry a single data value, never subfields.
        # A control tag arriving with subfields signals a corrupt upstream export.
        if v.startswith("00") and v[2].isdigit():
            raise ValueError(f"Tag {v!r} is a control field and must not contain subfields")
        return v


class BibliographicRecord(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    record_id: str = Field(alias="001", min_length=1)
    leader: str = Field(min_length=24, max_length=24)
    fields: list[MarcField]
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @field_validator("leader")
    @classmethod
    def enforce_static_leader_bytes(cls, v: str) -> str:
        # Positions 10-11 (indicator/subfield code counts) are always "22" in
        # valid MARC21; 20-23 is the fixed "4500" entry map. Deviations mean a
        # non-standard directory that the rest of the pipeline cannot parse.
        if v[10:12] != "22" or v[20:24] != "4500":
            raise ValueError("Leader static positions (10-11, 20-23) are non-conforming")
        return v

Step 2 — Validate at the ingestion boundary with early rejection

The parse stage hands the validator raw dictionaries. At this boundary, wrap construction in explicit exception handling so a single malformed record never halts the batch — reject it, preserve the original payload with a machine-readable reason, and route it to a quarantine queue. Preserving the original alongside the structured error is what makes automated reprocessing possible once an upstream vendor feed corrects its export.

import structlog
from pydantic import ValidationError

from .models import BibliographicRecord  # from Step 1
from .quarantine import QuarantineSink

log = structlog.get_logger()


def validate_boundary(
    raw_record: dict,
    quarantine: QuarantineSink,
    correlation_id: str,
) -> BibliographicRecord | None:
    """Structural gate. Returns an accepted model or None (record quarantined)."""
    try:
        record = BibliographicRecord(**raw_record)
    except ValidationError as exc:
        quarantine.send(
            original=raw_record,
            errors=exc.errors(),
            tier="structural",
            correlation_id=correlation_id,
        )
        log.warning(
            "schema.reject",
            tier="structural",
            correlation_id=correlation_id,
            error_count=len(exc.errors()),
        )
        return None
    return record

Pitfall: never catch a bare Exception here. ValidationError is recoverable and belongs in quarantine; a MemoryError or a broker ConnectionError is an infrastructure fault that must propagate so the orchestrator can retry the whole task rather than silently discarding records as if they were malformed.

Step 3 — Layer semantic validation over the structural pass

Structural clearance does not mean a record is correct — only that it is shaped correctly. The semantic tier cross-references controlled vocabularies and authority IDs, which requires I/O and therefore lives outside the Pydantic model. Keep it stateless and cache the authority lookups so the tier stays idempotent and horizontally scalable.

from dataclasses import dataclass
from functools import lru_cache

from .authority import AuthorityStore
from .models import BibliographicRecord

_KNOWN_THESAURI = frozenset({"0", "1", "2", "3", "4", "7"})  # 650 2nd indicator


@dataclass(frozen=True)
class SemanticIssue:
    tag: str
    subfield: str
    reason: str


@lru_cache(maxsize=100_000)
def _authority_exists(store: AuthorityStore, heading: str) -> bool:
    return store.contains(heading)


def semantic_review(
    record: BibliographicRecord,
    store: AuthorityStore,
) -> list[SemanticIssue]:
    """Non-blocking tier: returns advisory issues; record is still accepted."""
    issues: list[SemanticIssue] = []
    for field in record.fields:
        if field.tag != "650":
            continue
        if field.indicator2 not in _KNOWN_THESAURI:
            issues.append(SemanticIssue("650", "ind2", "unknown thesaurus source"))
        for sub in field.subfields:
            if sub.code == "a" and not _authority_exists(store, sub.value):
                issues.append(SemanticIssue("650", "a", f"unresolved heading {sub.value!r}"))
    return issues

Semantic issues are advisory: a subject heading that fails authority lookup should not block a title from reaching patrons, it should flag the record for cataloger review. Treating semantic failures as hard rejections is how pipelines end up quarantining thousands of perfectly displayable records over a single unmapped local thesaurus.

PII & Compliance Checkpoints

Public-sector catalog metadata routinely carries data that must never be persisted or forwarded to analytics: patron annotations in 500/583 local notes, restricted-access URLs in 856, and internal tracking codes in 9xx fields. The validator is the correct place to enforce data minimization, because it is the last stage that sees the full record before it fans out. This is the same discipline applied at export time in PII Masking in Patron Data Exports, pulled upstream to the ingestion boundary; the governing boundaries are set out in Data Privacy Boundaries in Library Systems.

Implement field-level redaction as a Pydantic serializer so masking happens automatically on every serialization path, with deterministic hashing that preserves an audit trail without retaining the raw value.

import hashlib
import os
import re

from pydantic import model_serializer

from .models import BibliographicRecord

_SALT = os.environ["PII_HASH_SALT"].encode()
_PII_PATTERNS = (
    re.compile(r"(?:patron|staff|internal)\s*id[:\s]+(\w+)", re.IGNORECASE),
    re.compile(r"(?:ssn|dob|phone)[:\s]+([\d\-]+)", re.IGNORECASE),
)


def _fingerprint(value: str) -> str:
    return hashlib.sha256(_SALT + value.encode()).hexdigest()[:12]


class SanitizedBibliographicRecord(BibliographicRecord):
    @model_serializer(mode="wrap")
    def mask_pii(self, handler):
        serialized = handler(self)
        for key, value in serialized.items():
            if not isinstance(value, str):
                continue
            for pattern in _PII_PATTERNS:
                value = pattern.sub(
                    lambda m: f"[REDACTED:{_fingerprint(m.group(1))}]",
                    value,
                )
            serialized[key] = value
        return serialized

Masking must be idempotent so a reprocessed record produces the identical fingerprint and never double-redacts. Retention flags belong here too: records carrying circulation-linked notes should be tagged with the retention class defined in Data Retention Policies for Public Libraries so downstream purge jobs can honor state schedules. Audit logging requires structured, immutable event streams: every validation pass emits a JSON log entry with a correlation ID, the tier that ran, pass/fail status, and masked payload fingerprints. Log schema hashes, field counts, and durations — never raw payloads — so logs stay queryable by SIEM platforms and compliant with NIST SP 800-53 audit controls.

Error Handling & Quarantine Patterns

Quarantine is not an error log; it is a durable, replayable holding area. The sink must persist the original payload, the structured error, the tier that rejected it, and a correlation ID, so an operator can triage in bulk and reprocess once the upstream feed is corrected. The message broker and dead-letter mechanics are governed by Async Batch Processing for Catalog Updates; the validator only needs a thin, well-typed sink.

import json
from datetime import datetime, timezone

import structlog

log = structlog.get_logger()


class QuarantineSink:
    """Durable sink for rejected records. Wraps a broker publisher."""

    def __init__(self, publisher, max_retries: int = 3) -> None:
        self._publisher = publisher
        self._max_retries = max_retries

    def send(
        self,
        *,
        original: dict,
        errors: list[dict],
        tier: str,
        correlation_id: str,
    ) -> None:
        envelope = json.dumps(
            {
                "original": original,
                "errors": errors,
                "tier": tier,
                "correlation_id": correlation_id,
                "quarantined_at": datetime.now(timezone.utc).isoformat(),
            }
        ).encode()

        last_exc: Exception | None = None
        for attempt in range(1, self._max_retries + 1):
            try:
                self._publisher.publish(envelope)
                return
            except ConnectionError as exc:  # transient broker fault
                last_exc = exc
                log.warning("quarantine.retry", attempt=attempt, correlation_id=correlation_id)
        # Exhausted retries: raise so the orchestrator can dead-letter the whole task
        # rather than losing a record silently.
        raise RuntimeError(f"quarantine publish failed after {self._max_retries} attempts") from last_exc

Distinguish the two failure classes carefully. A ValidationError means the record is bad and belongs in quarantine. A ConnectionError reaching the sink means the infrastructure is bad; retry with backoff and, if it persists, raise so the task dead-letters and is retried whole. Silently swallowing the second class is how records vanish between the vendor feed and the catalog with no trace.

Performance Considerations

Validation runs on every record in every batch, so its per-record cost sets the ceiling on ingestion throughput. Three levers matter. First, keep the validator stateless and idempotent so it scales horizontally — validation workers consume from the broker and apply checks in parallel, with no shared mutable state between records. Second, cache authority lookups aggressively; the lru_cache in Step 3 turns a per-record network round trip into an amortized in-process hit, and a warm cache is the single largest throughput win on semantically dense record sets. Third, do not materialize whole batches — validate a stream of records and emit each accepted delta immediately, retaining nothing across iterations. When a source file is large enough that even parsing it would exhaust the worker, the streaming techniques in Optimizing pymarc Performance for Large Record Sets apply upstream, and the validator must preserve that discipline by never collecting results into a list.

Where the pipeline pulls records from a live ILS rather than a batch drop, validation throughput has to be paced against the acquisition rate governed by ILS REST API Polling & Rate Limiting; a validator faster than the feed simply idles, so size the worker pool to the sustained ingest rate, not the peak.

Verification & Testing

Prove each tier independently. Unit-test the structural contract by constructing records that violate exactly one rule and asserting the specific ValidationError location; test the semantic tier against a mocked authority store so it needs no live database; and test the quarantine sink with a publisher stub that fails a set number of times to confirm retry and dead-letter behavior.

import pytest
from pydantic import ValidationError

from .models import BibliographicRecord, MarcField


def _valid_record() -> dict:
    return {
        "001": "ocm123456",
        "leader": "00000nam a2200000 a 4500",
        "fields": [{"tag": "245", "indicator1": "1", "indicator2": "0",
                    "subfields": [{"code": "a", "value": "A valid title"}]}],
    }


def test_accepts_conforming_record():
    record = BibliographicRecord(**_valid_record())
    assert record.record_id == "ocm123456"


def test_rejects_control_field_with_subfields():
    bad = _valid_record()
    bad["fields"] = [{"tag": "001", "subfields": [{"code": "a", "value": "x"}]}]
    with pytest.raises(ValidationError) as exc_info:
        BibliographicRecord(**bad)
    assert "control field" in str(exc_info.value)


def test_forbids_unexpected_vendor_extension():
    bad = _valid_record()
    bad["vendor_local_flag"] = "Z"
    with pytest.raises(ValidationError) as exc_info:
        BibliographicRecord(**bad)
    assert "extra" in str(exc_info.value).lower()


def test_masking_is_idempotent():
    from .sanitize import SanitizedBibliographicRecord
    raw = _valid_record()
    raw["fields"][0]["subfields"][0]["value"] = "note patron id: 998877"
    once = SanitizedBibliographicRecord(**raw).model_dump()
    twice = SanitizedBibliographicRecord(**once | {"001": raw["001"]}).model_dump()
    assert once == twice or "[REDACTED:" in str(once)

Run the suite in CI against a corpus of real anonymized vendor records, not just synthetic fixtures — the failure modes that matter (control fields with stray subfields, MARC-8 encoding masquerading as UTF-8, headings against retired thesauri) come from real exports and rarely appear in hand-written tests.

Troubleshooting & FAQ

Records that parse fine are rejected by the validator with an extra error. Why?

extra="forbid" on the model rejects any key not declared on BibliographicRecord. A vendor feed shipping local 9xx fields or a bespoke flag will trip it. This is intentional — it surfaces schema drift instead of silently absorbing it. Either add the field to the model with an explicit constraint, or strip vendor-local keys in a normalization step handled by ILS Schema Translation Patterns before validation, so the contract stays a deliberate allowlist.

The whole batch aborts when one record is malformed. How do I isolate failures?

The construction call is not wrapped. Move each record through validate_boundary (Step 2), which catches ValidationError per record and routes it to quarantine while the batch continues. Never wrap the whole batch in a single try/except — that discards every record after the first failure.

Diacritics validate but arrive as mojibake (e.g. Müller) in the catalog. What went wrong?

The record was MARC-8 encoded (leader position 09 = space) but decoded as UTF-8 upstream. This is an encoding fault the syntactic tier should catch, but the fix belongs at parse time — see Handling UTF-8 Encoding in Legacy MARC Records. Add a syntactic check that rejects records whose declared coding scheme does not match the decoded byte content.

Thousands of records are quarantined over unresolved subject headings. Is that right?

No — that is a tier misconfiguration. Authority-lookup failures are semantic and advisory; they should be returned as SemanticIssue flags for cataloger review, not raised as ValidationError. Only syntactic and structural violations block a record. Re-check that semantic_review collects issues and returns them rather than raising.

After upgrading Pydantic, the masking serializer and validators stop working. What changed?

Pydantic v2 replaced v1’s @validator with @field_validator, @root_validator with @model_validator, and introduced @model_serializer. Code written for v1 will not run. Pin pydantic>=2.5,<3 and migrate the decorators; the models and serializer on this page assume the v2 API throughout.

Quarantined records disappear and never come back. How do I recover them?

The quarantine sink is dropping payloads instead of persisting them durably, or the broker publish is failing silently. Confirm QuarantineSink.send raises on exhausted retries (so the task dead-letters) rather than swallowing the exception, and that the quarantine queue is durable. Replay is only possible if the original payload was preserved with its correlation ID.

Related

• Validating MARC Leader Fields Before Database Insert — the byte-level leader gate this page’s structural tier delegates to
• Parsing MARC Records with pymarc — the parse stage that hands structured records to the validator
• Async Batch Processing for Catalog Updates — broker, dead-letter, and reprocessing mechanics behind the quarantine queue
• PII Masking in Patron Data Exports — the masking discipline this page pulls upstream to the ingestion boundary
• MARC21 Field Mapping for Modern Pipelines — translating validated tags into the canonical internal model