Operational Context

A Python automation engineer loads a vendor-delivered Dublin Core (DC) manifest into CollectionBase, the batch runs for twenty minutes, and the ingestion API returns a wall of 422 Unprocessable Entity responses on the final commit — after the harvest has already consumed the memory budget. A collections manager, meanwhile, has no way to see which of the fourteen thousand records failed or why. Both symptoms trace to the same gap: there is no deterministic validation gate between the flat DC payload and CollectionBase’s relational schema, so type mismatches surface only at the network boundary. This page specifies that gate — a memory-bounded pipeline that transforms repeatable, untyped DC elements into strictly typed CollectionBase records and routes every rejection to a triageable manifest. It sits directly under the LIDO-to-database mapping layer, which owns the broader XML-to-database routing this validation feeds, and inherits the structural conventions of the Core Architecture & Collection Taxonomy baseline.

Root Cause Analysis

Dublin Core and CollectionBase encode incompatible contracts, and the mismatch is invisible until commit time.

  • Cardinality mismatch. DC prioritises interoperability through minimalism: dc:creator, dc:subject, and dc:contributor are all unbounded and repeatable. CollectionBase enforces strict cardinality — one primary title, one normalised creator string — so naive parsing that overwrites on each repeated element silently drops all but the last value.
  • Untyped coercion. DC permits a bare dc:date of c. 1890 or 19th century. CollectionBase requires ISO 8601. Unvalidated strings pass every local str check and only fail when the database constraint rejects them mid-transaction, which the API surfaces as a 422.
  • Missing vocabulary binding. Free-text dc:subject strings carry no controlled-vocabulary identity. Without resolution against an authority file, the same concept enters the schema under a dozen spellings.
  • Namespace and memory faults. Prefixed elements resolved against the wrong namespace URI vanish silently, and loading the entire DOM into memory triggers OOM kills on multi-gigabyte manifests — the same failure covered in handling large CSV batches without memory leaks.

The fix shifts validation left: a deterministic mapping contract that rejects each violation at the layer that owns it, before any record reaches the relational schema.

Pipeline Architecture & Data Flow

The validation layer operates as a streaming transformer. Raw XML enters a memory-optimised iterator; parsed nodes pass through a Pydantic schema validator; controlled-vocabulary terms resolve against a local cache; validated records serialise to CollectionBase-compatible JSON; invalid records route to a structured error manifest. Memory overhead stays constant regardless of payload size.

Streaming Dublin Core validation pipeline Dublin Core XML feeds a streaming iterparse reader, then a namespace-resolve and accumulate stage, then a Pydantic validation decision on ISO 8601 dates and URIs. The invalid branch drops down to a structured error manifest; the valid branch flows right into Getty AAT and TGN vocabulary binding and then to CollectionBase JSON output. valid invalid Dublin Core XML flat · untyped Stream records iterparse · one record Namespace resolve accumulate repeats into lists Pydantic ISO 8601 · URIs Vocabulary binding Getty AAT / TGN CollectionBase JSON strictly typed Error manifest index · id · reason

Canonical Solution

The pipeline is four cooperating stages: stream, enforce, bind, dispatch. Each stage owns one contract and rejects violations locally.

Stage 1 — Streaming ingestion and namespace resolution

Replace bulk DOM loading with lxml.etree.iterparse. Match records by their fully qualified namespace tag so prefixed elements cannot be dropped, and accumulate repeatable elements into lists rather than overwriting — this is the cardinality contract, handled before Pydantic ever sees the record.

python
from __future__ import annotations
from lxml import etree
from typing import Iterator

DC_NS = "http://purl.org/dc/elements/1.1/"

def stream_dc_records(xml_path: str) -> Iterator[dict[str, list[str]]]:
    # iterparse fires on element end, so peak memory is one record, not the file.
    context = etree.iterparse(xml_path, events=("end",), tag=f"{{{DC_NS}}}record")
    for _, elem in context:
        record: dict[str, list[str]] = {}
        for child in elem:
            if not child.text or not child.text.strip():
                continue
            key = child.tag.split("}")[-1]  # strip {namespace} -> local name
            # DC elements are repeatable (multiple dc:creator), so accumulate
            # into a list rather than overwriting the previous value.
            record.setdefault(key, []).append(child.text.strip())
        elem.clear()  # release the parsed subtree
        while elem.getprevious() is not None:
            del elem.getparent()[0]  # drop already-processed siblings
        yield record

Stage 2 — Schema enforcement and type coercion

Pydantic v2 models enforce the type and cardinality contract. Field validators run at construction time, so an ambiguous date or a non-repeatable field with two values is rejected before serialisation — the same strict-model discipline described in schema validation with Pydantic and the record contracts in designing museum object schemas.

python
from pydantic import BaseModel, field_validator
import re

class CollectionBaseRecord(BaseModel):
    model_config = {"str_strip_whitespace": True, "extra": "forbid"}

    object_id: str
    title_primary: str
    creator_normalized: str
    date_created: str
    subject_uris: list[str]

    @field_validator("date_created")
    @classmethod
    def validate_iso8601(cls, v: str) -> str:
        # CollectionBase indexes dates for range queries; only ISO 8601 admitted.
        if not re.match(r"^\d{4}(-\d{2}){0,2}$", v):
            raise ValueError("Requires YYYY, YYYY-MM, or YYYY-MM-DD format")
        return v

    @field_validator("subject_uris")
    @classmethod
    def enforce_uri_format(cls, v: list[str]) -> list[str]:
        if not all(uri.startswith(("http://", "https://")) for uri in v):
            raise ValueError("Subjects must resolve to absolute URIs")
        return v

Stage 3 — Vocabulary binding and URI resolution

Free-text DC subjects require normalisation against Getty vocabularies before they can satisfy subject_uris. A cached resolver maps each string to an AAT identifier and rejects unresolvable terms — the identifier-resolution mechanics live in implementing Getty AAT & TGN.

python
from functools import lru_cache
import json
import requests

@lru_cache(maxsize=2048)
def resolve_aat_uri(term: str) -> str:
    endpoint = "https://vocab.getty.edu/sparql.json"
    # Bind the term via json.dumps so values containing apostrophes
    # (e.g. O'Keeffe) cannot break or inject the SPARQL query.
    query = (
        "PREFIX skos: <http://www.w3.org/2004/02/skos/core#> "
        "SELECT ?uri WHERE { ?uri skos:prefLabel ?label . "
        f"FILTER(STR(?label) = {json.dumps(term)} && LANG(?label) = 'en') }}"
    )
    response = requests.get(endpoint, params={"query": query}, timeout=10)
    response.raise_for_status()
    bindings = response.json().get("results", {}).get("bindings")
    if not bindings:
        raise ValueError(f"Unresolvable term: {term}")
    return bindings[0]["uri"]["value"]

Stage 4 — Batch dispatch and error triage

A generator drives the pipeline end to end. Validated records yield straight to the ingestion queue; ValidationError exceptions capture into a structured manifest a collections manager can act on, eliminating manual reconciliation.

python
from pydantic import ValidationError
import csv, sys

def validate_and_dispatch(records: Iterator[dict[str, list[str]]]) -> int:
    failures = csv.writer(sys.stderr)
    ok = 0
    for idx, raw in enumerate(records):
        try:
            record = CollectionBaseRecord(
                object_id=raw["identifier"][0],
                title_primary=raw["title"][0],
                creator_normalized="; ".join(raw.get("creator", [])),
                date_created=raw["date"][0],
                subject_uris=[resolve_aat_uri(s) for s in raw.get("subject", [])],
            )
            yield_to_queue(record.model_dump())  # your ingestion sink
            ok += 1
        except (ValidationError, ValueError, KeyError, IndexError) as e:
            failures.writerow([idx, raw.get("identifier", ["?"])[0], str(e)])
    return ok

Edge Cases and Variants

  • CSV vs. XML input. Vendors that ship DC as a flat CSV skip Stage 1 entirely — feed csv.DictReader rows into a shim that wraps each single-valued cell in a list, so Stages 2–4 stay identical. Pipe-delimited multi-value cells split on | before wrapping.
  • OAI-PMH API payloads. When DC arrives from an OAI-PMH endpoint rather than a file, replace iterparse(path) with iterparse(BytesIO(response.content)) and resume from the resumptionToken on each page.
  • Strict vs. lenient mode. For provisional harvests, relax the date validator to coerce c. 1890 to 1890 with a logged warning instead of raising; keep strict mode (extra: "forbid", raise-on-ambiguity) for production commits.
  • Alternative vocabularies. Swap resolve_aat_uri for a TGN resolver on dc:coverage place terms, or point the SPARQL endpoint at a local triplestore mirror to remove the network dependency and the 10-second timeout.
  • Qualified Dublin Core. dcterms: refinements such as dcterms:created carry their own namespace; add the DCTERMS URI to the tag matcher rather than folding them into unqualified dc:date.

Validation

Confirm the gate accepts a clean record and rejects an ambiguous date before wiring it into the ingestion queue.

python
import pytest
from pydantic import ValidationError

def test_iso8601_gate_accepts_and_rejects():
    good = CollectionBaseRecord(
        object_id="obj-001",
        title_primary="Untitled",
        creator_normalized="O'Keeffe, Georgia",
        date_created="1929-05",
        subject_uris=["http://vocab.getty.edu/aat/300033618"],
    )
    assert good.date_created == "1929-05"

    with pytest.raises(ValidationError):
        CollectionBaseRecord(
            object_id="obj-002",
            title_primary="Untitled",
            creator_normalized="Unknown",
            date_created="c. 1890",       # not ISO 8601 -> rejected here
            subject_uris=["paintings"],   # not an absolute URI -> also rejected
        )

Run it with pytest -q test_validation.py; a green run proves the contract fails ambiguous input at construction time rather than at the API boundary.

Standards Alignment

The gate keeps DC records interoperable across the standards this pipeline touches. subject_uris binds free text to Getty AAT identifiers, so federated queries resolve the same concept regardless of source spelling, and the same resolver serves TGN place terms. ISO 8601 date_created values remain sortable for the temporal filtering that discovery interfaces and IIIF Presentation API 3.0 range navigation depend on, and a stable object_id maps cleanly to a canonical IIIF manifest URI. Because these records descend from the same routing layer that ingests LIDO, the typed output stays semantically aligned with the museum’s LIDO-derived records — the cardinality and vocabulary contracts enforced here are exactly the ones LIDO assumes downstream. Element definitions follow the Dublin Core Metadata Initiative specifications.

FAQ

Why does my Dublin Core batch pass local checks but fail with 422 at commit?

The DC payload is valid XML and every field is a valid str, so nothing fails locally — but CollectionBase enforces ISO 8601 dates, strict cardinality, and URI-bound subjects that a bare string check never verifies. The constraint fires inside the database transaction, which the API surfaces as 422 Unprocessable Entity. Running the Pydantic gate before dispatch moves each rejection to construction time with an actionable message.

How do I keep repeatable dc:creator values from being silently dropped?

Accumulate every repeated element into a list during Stage 1 (record.setdefault(key, []).append(...)) rather than assigning, then decide the CollectionBase mapping explicitly — join multiple creators into one normalised string, or promote them to a related-agents table. Overwriting on each element keeps only the last value and loses the rest without any error.

How is the pipeline memory-bounded on multi-gigabyte manifests?

lxml.etree.iterparse yields one record at a time and the loop calls elem.clear() plus deletes processed siblings, so peak memory is a single record plus the vocabulary cache — not the whole file. This is what prevents the OOM kill that whole-DOM parsing triggers on large harvests.

Can I validate qualified Dublin Core (dcterms:) with the same gate?

Yes. Add the DCTERMS namespace URI to the tag matcher so refinements like dcterms:created and dcterms:spatial are captured under their own keys instead of being folded into unqualified elements, then map them to the appropriate CollectionBaseRecord fields. The validators themselves are unchanged.

What happens when a Getty AAT term will not resolve?

resolve_aat_uri raises ValueError for any term with no English skos:prefLabel match, which Stage 4 catches and writes to the error manifest with the record index and offending term. The record is quarantined rather than committed with an unbound subject, so a curator can supply the correct authority term and re-run only the failed rows.