Workflow Context

Museum ingestion pipelines take in heterogeneous manifests — legacy TMS exports, donor spreadsheets, digitization-workstation dumps — and every one of them arrives with its own idea of what a valid date, a rights code, or an asset path looks like. Without a strict enforcement point, a malformed creation_date, a missing rights flag, or an asset URI pointing at a defunct file share slips through and silently corrupts the collection management system. This page specifies the enforcement point: a Pydantic v2 model that performs runtime type checking, constraint validation, legacy-column aliasing, and structured error reporting before a single record reaches a database transaction.

Within the parent stage — Automated Record Ingestion & Sync Workflows — this is the validation gate that splits the stream. It runs after a payload leaves the message broker and before any write, so malformed records divert to quarantine while conforming records advance to normalization. Every other workflow in that section routes its output through the model described here: the CSV chunks produced by CSV to database sync strategies, the API pages pulled by building async ingestion pipelines, and the fields lifted from scans by automating OCR metadata extraction all become the same validated shape at this boundary. The role that owns this code is the Python automation engineer; the role that consumes its output — the structured rejection reports — is the collections manager reconciling a drifted export.

Prerequisites

Before building the validation gate, confirm the following are in place:

  • Python 3.9+ for PEP 604 union syntax (str | None) and native generic type hints used throughout the examples below.
  • Pydantic v2 (pip install "pydantic>=2.6") — the v1 @validator and class Config APIs are not compatible with the field_validator / model_config patterns here. See the official Pydantic v2 documentation for the migration surface.
  • A canonical field contract for your institution — the indexed columns and archival blob defined by designing museum object schemas. The model is only as correct as the schema it mirrors.
  • The LIDO 1.1 element vocabulary you validate against, so field names and cardinality match the interchange envelope; consult the LIDO schema for element definitions.
  • A RightsStatements.org controlled-vocabulary list to constrain the rights_statement field, resolved through the rules in mapping RightsStatements.org to collection fields.
  • A thread pool executor and an async event loop if you validate in batch — CPU-bound validation must be offloaded so it does not stall the ingestion loop.
  • A quarantine table or dead-letter store with columns for the raw payload, the structured error list, and a UTC timestamp.

Schema Reference

The model mirrors the institutional metadata contract one field at a time. Each field carries an explicit type, a constraint, and — where the source diverges from canonical naming — an alias that bridges a legacy column to the internal identifier without silent coercion. The table below is the specification the code enforces.

Field Type Constraint Source / vocabulary
accession_number str regex ^[A-Z]{2,4}\.\d{4}\.\d{1,4}$ Institutional accession scheme
title str min_length=2, max_length=255 LIDO titleSet
creator str | None optional LIDO nameActorSet; later resolved to a Getty ULAN URI
creation_date date | None ISO 8601 or legacy year coerced before typing LIDO eventDate
rights_statement str required, aliased from rights_code RightsStatements.org URI
asset_uris list[str] each must match an allowed scheme Institutional DAM routing rules
checksum str | None regex ^[a-f0-9]{64}$ (SHA-256) Fixity manifest

Define the model with strict typing and unknown-field rejection so a drifted export fails loudly rather than writing partial records. Free-text creator and materials values are validated for shape here, then resolved to stable authority URIs downstream — see implementing Getty AAT & TGN for that resolution step.

python
from datetime import date, datetime, timezone
from pydantic import BaseModel, ConfigDict, Field, field_validator, ValidationError
import logging

logger = logging.getLogger("museum_pipeline.validation")


class DigitalAssetRecord(BaseModel):
    """Institutional object contract enforced at the ingestion gate."""
    model_config = ConfigDict(strict=True, extra="forbid", populate_by_name=True)

    accession_number: str = Field(pattern=r"^[A-Z]{2,4}\.\d{4}\.\d{1,4}$")
    title: str = Field(min_length=2, max_length=255)
    creator: str | None = None
    creation_date: date | None = None
    rights_statement: str = Field(alias="rights_code")
    asset_uris: list[str] = Field(default_factory=list)
    checksum: str | None = Field(default=None, pattern=r"^[a-f0-9]{64}$")

strict=True blocks Pydantic’s default coercion — an integer accession_number or a stringified boolean is rejected instead of quietly cast. extra="forbid" rejects any payload carrying an unmapped column from a drifted export. populate_by_name=True lets the model accept either the legacy rights_code alias or the canonical rights_statement field name, which matters when the same model validates both a raw legacy export and an already-normalized internal record.

Step-by-Step Implementation

1. Normalize inputs with mode="before" validators

Legacy museum data carries inconsistent date formats and non-standard URI schemes. A field_validator running in mode="before" sees the raw input ahead of type coercion, so it can parse ISO 8601 variants and legacy year-only strings into a real date before strict typing rejects the string. These validators live on the DigitalAssetRecord class defined above.

python
    @field_validator("creation_date", mode="before")
    @classmethod
    def parse_iso_or_legacy_date(cls, v: object) -> object:
        if isinstance(v, str):
            for fmt in ("%Y-%m-%d", "%Y-%m", "%Y"):
                try:
                    return datetime.strptime(v, fmt).date()
                except ValueError:
                    continue
            raise ValueError(f"Unrecognized date format: {v}")
        return v

    @field_validator("asset_uris")
    @classmethod
    def validate_dam_paths(cls, v: list[str]) -> list[str]:
        valid_schemes = ("https://", "s3://", "file://")
        for uri in v:
            if not uri.startswith(valid_schemes):
                raise ValueError(f"Invalid asset URI scheme: {uri}")
        return v

The date validator tries formats in descending specificity; a 1923 accession-card entry becomes date(1923, 1, 1) while a malformed string raises a ValueError that Pydantic wraps into a field-level ValidationError. The URI validator runs after typing (default mode="after"), so it only sees a well-typed list[str] and can assert institutional DAM routing rules. Input variants: a CSV cell delivers asset_uris as a delimited string, so a CSV adapter must split it into a list before validation; an XML or JSON-API payload usually delivers it as a real array. Keep that splitting in the source adapter, not the model, so the contract stays transport-agnostic.

2. Validate a single record and capture structured errors

model_validate is the entry point. On failure it raises a ValidationError whose .errors() method returns a machine-readable list of field, rule, and offending input — the raw material for a quarantine record.

python
def validate_one(raw: dict) -> DigitalAssetRecord | dict:
    try:
        return DigitalAssetRecord.model_validate(raw)
    except ValidationError as exc:
        return format_validation_error(exc, raw)

3. Offload batch validation to a thread pool

Large manifests exceed memory when loaded synchronously, and validation is CPU-bound, so running it inline stalls the async ingestion loop. Stream records through an async generator and offload each model_validate call to a thread pool executor. Compliant records yield straight to the sync layer; failures are logged and skipped. For executor-pool sizing against the database connection pool, see building async ingestion pipelines.

python
import asyncio
from concurrent.futures import ThreadPoolExecutor
from collections.abc import AsyncGenerator

executor = ThreadPoolExecutor(max_workers=4)


async def validate_batch(
    records: list[dict],
) -> AsyncGenerator[DigitalAssetRecord, None]:
    loop = asyncio.get_running_loop()
    tasks = [
        loop.run_in_executor(executor, DigitalAssetRecord.model_validate, record)
        for record in records
    ]
    for completed in asyncio.as_completed(tasks):
        try:
            yield await completed
        except ValidationError as exc:
            logger.warning("Validation failure: %s", exc.errors(include_url=False))
            continue

as_completed yields records in completion order rather than submission order, which maximizes throughput but means the consumer must not assume input ordering. If downstream commit logic depends on deterministic ordering — for example a deduplication window keyed on accession number — collect results into a list and sort before committing instead of yielding eagerly.

4. Format the rejection into a quarantine payload

Structured error reporting is what makes compliance routing deterministic. Map each ValidationError to a quarantine record carrying the original payload, the field-level error list, and a UTC timestamp, so collections staff can review failures with the accession number attached rather than a stack trace.

python
def format_validation_error(error: ValidationError, raw: dict) -> dict:
    detail = error.errors(include_url=False)
    return {
        "status": "rejected",
        "accession_number": raw.get("accession_number"),
        "errors": detail,
        "input_data": [e.get("input") for e in detail],
        "quarantined_at": datetime.now(timezone.utc).isoformat(),
    }

Route the returned dict to the quarantine table. Because the raw payload is preserved intact, a corrected export can be re-ingested and re-validated without reconstructing the original input — the audit trail is complete.

Validation Flow

The gate is a simple split: raw records fan out to thread-pool validation, and each result routes on a single valid/invalid decision.

Pydantic validation gate — the valid/invalid split Left to right: raw manifest records enter a strict DigitalAssetRecord.model_validate call offloaded to a thread pool. A single Valid? decision splits the stream. Conforming records branch up and yield to the downstream sync layer; malformed records branch down through format_validation_error into the quarantine table, which preserves the raw payload, the structured field-level error list, and a UTC timestamp for later re-ingestion. One split — every record routes on a single valid/invalid decision Raw manifest records heterogeneous DigitalAssetRecord model_validate thread pool · strict Valid? Yield to sync layer conforming record advances to normalize format_validation_error → quarantine table raw payload preserved error list · UTC timestamp yes no

Rights and Access Routing

Validation is also where the rights gate first bites. The rights_statement field is required and — in production — should be constrained to the RightsStatements.org controlled vocabulary, not merely typed as a string. A record that validates its shape but carries a rights value outside the vocabulary is worse than a rejection, because it commits with a plausible-looking but unenforceable rights assertion. Tighten the field to a regex that pins the URI namespace, or validate it against a resolved vocabulary set at ingest time.

Crucially, this gate asserts only that a machine-readable rights statement is present and well-formed. The policy it implies — whether the asset is embargoed, watermarked, or eligible for a public replica — is resolved downstream in rights metadata mapping and licensing automation, and time-based restrictions are applied by implementing embargo workflows. Keeping the validation model ignorant of publication policy is deliberate: it means a bulk re-ingest can re-run this gate safely without re-deriving access decisions or accidentally re-exposing restricted material. The model guarantees a valid rights URI exists; the rights layer decides what that URI permits.

Verification and Testing

Prove the gate before it protects production. Assert-based tests cover the three cases that matter: a clean record validates, a legacy date coerces correctly, and a drifted or malformed payload is rejected with a usable error.

python
def test_valid_record_passes():
    rec = DigitalAssetRecord.model_validate({
        "accession_number": "PMA.1998.42",
        "title": "Untitled study",
        "creation_date": "1923",  # legacy year-only
        "rights_code": "https://rightsstatements.org/vocab/InC/1.0/",
        "asset_uris": ["https://dam.example.org/pma-1998-42.tif"],
    })
    assert rec.creation_date == date(1923, 1, 1)
    assert rec.rights_statement.endswith("/InC/1.0/")


def test_drifted_column_is_rejected():
    import pytest
    with pytest.raises(ValidationError) as exc_info:
        DigitalAssetRecord.model_validate({
            "accession_number": "PMA.1998.42",
            "title": "Untitled study",
            "rights_code": "https://rightsstatements.org/vocab/InC/1.0/",
            "legacy_curator_note": "loose column from a drifted export",
        })
    assert any(e["type"] == "extra_forbidden" for e in exc_info.value.errors())


def test_bad_uri_scheme_quarantines():
    payload = {
        "accession_number": "PMA.1998.42",
        "title": "Untitled study",
        "rights_code": "https://rightsstatements.org/vocab/InC/1.0/",
        "asset_uris": ["ftp://legacy-share/pma.tif"],  # disallowed scheme
    }
    result = validate_one(payload)
    assert isinstance(result, dict)
    assert result["status"] == "rejected"

Run with pytest -q; all three cases should pass. For a pre-flight check on a new export, add a dry-run mode that streams a full manifest through validate_batch, counts committed versus quarantined records, and prints the quarantine reasons without writing anything — this surfaces a drifted export before it touches the catalogue.

In This Section

This validation model is the shared contract the rest of the ingestion stage depends on, so its deepest treatments live in the sibling workflows that exercise it against real inputs:

Standards Alignment

The validated record maps cleanly onto interchange and delivery standards. Its fields mirror LIDO 1.1 element definitions — title to titleSet, creation_date to eventDate, creator to nameActorSet — so a conforming record flattens into the LIDO envelope without lossy re-shaping, per mapping LIDO to internal databases. The rights_statement URI is asserted before any record is publish-eligible, satisfying the RightsStatements.org requirement that every published object carry a machine-readable rights value. Validated asset_uris and the rights_statement field together supply the inputs a downstream generator needs to build an IIIF Presentation API 3.0 manifest with a correct rights property. Strict validation is therefore not busywork — it is the precondition that lets every downstream standard be met without reconciliation.

FAQ

Why do valid-looking payloads still fail with extra_forbidden?

Because extra="forbid" rejects any field not declared on the model, and a drifted export often adds a stray column — a curator note, a legacy status flag — that was never mapped. This is intentional: the alternative is silently dropping the column and committing a record that no longer matches its source. Inspect exc.errors() for the extra_forbidden entry to see the offending field name, then either add it to the model or fix the export.

My legacy dates raise Unrecognized date format. How do I handle them?

The mode="before" date validator only tries %Y-%m-%d, %Y-%m, and %Y. Archival card data often carries date ranges (“circa 1920–1925”) or descriptive text (“late 19th century”) that no strptime format matches. Route those to a separate date_display string field and leave creation_date null rather than forcing a false-precision value; the display string preserves the curatorial statement while the typed field stays clean for range queries.

Should I use strict=True for every field?

Use strict mode for the model as a default, but remember that mode="before" validators run ahead of strict typing — that is exactly how the date field accepts a legacy string and still ends up strictly typed as a date. Strict mode blocks Pydantic’s implicit coercion (string-to-int, int-to-bool); your explicit before validators are where controlled, auditable coercion belongs. Never disable strict mode globally to paper over one field’s parsing problem.

Why validate in a thread pool instead of inline?

Pydantic validation is CPU-bound. Running it inline inside an async ingestion loop blocks the event loop, so concurrent broker fetches and database writes stall behind it. Offloading each model_validate call to a ThreadPoolExecutor keeps the loop responsive. Size the pool to your CPU count, not your I/O concurrency — more threads than cores just adds context-switching overhead for CPU-bound work.

How do I stop one bad record from aborting the whole batch?

Wrap each await completed in its own try/except ValidationError inside the as_completed loop, as shown above, and continue on failure. Because each record is an independent executor task, a single rejection routes to quarantine while its siblings validate and yield normally — one malformed manifest row never takes down the batch.