Automated Record Ingestion & Sync Workflows

Museum collection management systems live or die on the reliability of the pipeline that feeds them. When a digitization campaign produces forty thousand new capture records in a weekend, or a nightly export from a legacy Text Management System (TMS) instance drifts by a single column, the ingestion layer is where the damage either stops or spreads. This is the stage that turns heterogeneous inputs — donor spreadsheets, archival XML, OCR output, object-storage events — into validated, deduplicated, standards-conformant records inside the primary repository, without ever blocking the curatorial staff who are querying that same database in real time.

Three institutional roles depend on getting this right. The collections manager needs deterministic outcomes: every record that entered the pipeline is either committed, quarantined with a readable reason, or explicitly rejected — never silently dropped or half-written. The Python automation engineer needs bounded concurrency and memory safety so a bulk run does not exhaust database connections or the digitization host’s RAM. The DAMS administrator needs an immutable audit trail linking every committed field back to its source payload for provenance and rollback. This page defines the reference architecture that satisfies all three, and indexes the deeper workflows that implement each stage. It sits alongside the core architecture and collection taxonomy that governs how records are modeled once they land, and hands its validated output to rights metadata mapping and licensing automation before anything reaches a public endpoint.

Architecture Overview #

Reliable ingestion runs on an event-driven model, not synchronous batch execution. Incoming payloads from digitization labs or archival exports route through a message broker that isolates network latency and parsing failures from the primary write path. Idempotency keys derived from accession numbers or IIIF manifest URIs prevent duplicate writes during retry cycles. Checkpointing tracks processing offsets which, combined with those idempotency keys, deliver effectively-once processing — brokers themselves guarantee only at-least-once delivery, so the application layer must supply the deduplication. A schema-validation gate splits the stream: malformed payloads divert to a dead-letter queue for forensic review, while conforming records proceed to transformation and a bulk upsert against the collection management system. The diagram below shows the full stage this page covers, from source ingress to committed record.

Every arrow in that diagram is an explicit gate rather than a best-effort handoff. The message broker gate absorbs burst pressure; the validation gate enforces institutional metadata standards; the upsert gate enforces idempotency at the database boundary. The sections that follow specify the standards each gate must honour, the canonical Python that wires the worker pool together, and the deeper implementations for each ingress format.

Standards Landscape #

Ingestion is not a neutral plumbing exercise — every record must arrive conformant to the heritage standards that downstream aggregation, discovery, and rights enforcement depend on. Validating against these specifications at ingest time is far cheaper than reconciling drift after thousands of records have already been committed. The table below maps each standard to the pipeline stage where it is enforced and the concrete role it plays in this workflow.

Version pinning matters here. IIIF 3.0 changed the manifest structure enough that a validator tuned for Presentation API 2.1 will accept payloads that later fail in a compliant viewer, so the ingestion gate must target the exact major version your delivery layer serves. Getty vocabularies are consumed as Linked Open Data rather than a versioned file, which means the normalization stage must cache resolved URIs and periodically reconcile deprecated terms — a concern addressed in the taxonomy layer’s guidance on implementing Getty AAT & TGN.

Core Implementation Pattern #

The canonical task for this pipeline stage is a bounded, idempotent batch coordinator: it pulls a batch of raw records off the broker, validates each against a strict Pydantic v2 model, routes failures to quarantine, and hands the clean batch to a bulk upsert. The pattern below is deliberately transport-agnostic — the same coordinator drives a CSV stream, a polled API cursor, or a broker subscription simply by swapping the source generator. Note the use of Python 3.9+ syntax (PEP 604 | unions) and the Pydantic v2 API (model_config, field_validator, model_dump).

import asyncio
import hashlib
import logging
from collections.abc import AsyncIterator
from datetime import date, datetime, timezone

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

logger = logging.getLogger("ingestion.coordinator")

MAX_CONCURRENCY = 16
BATCH_SIZE = 500


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

    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
    date_created: date | None = None
    rights_uri: str = Field(pattern=r"^https://rightsstatements\.org/vocab/")
    source_manifest: str | None = None

    @field_validator("title")
    @classmethod
    def collapse_whitespace(cls, v: str) -> str:
        return " ".join(v.split())

    def idempotency_key(self) -> str:
        # Stable key survives retries; manifest URI wins when present.
        seed = self.source_manifest or self.accession_number
        return hashlib.sha256(seed.encode("utf-8")).hexdigest()


async def validate_batch(
    raw_records: list[dict],
) -> tuple[list[ObjectRecord], list[dict]]:
    """Split a raw batch into (valid records, quarantine payloads)."""
    valid: list[ObjectRecord] = []
    quarantined: list[dict] = []
    for raw in raw_records:
        try:
            valid.append(ObjectRecord.model_validate(raw))
        except ValidationError as exc:
            quarantined.append({
                "payload": raw,
                "errors": exc.errors(include_url=False),
                "quarantined_at": datetime.now(timezone.utc).isoformat(),
            })
    return valid, quarantined


async def run_ingestion(
    source: AsyncIterator[list[dict]],
    commit,          # async callable: (list[ObjectRecord]) -> int
    quarantine,      # async callable: (list[dict]) -> None
) -> None:
    """Drive validation + commit with bounded concurrency."""
    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
    seen: set[str] = set()

    async def handle(batch: list[dict]) -> None:
        async with semaphore:
            valid, bad = await validate_batch(batch)
            # Deduplicate within and across batches before writing.
            deduped = [r for r in valid if r.idempotency_key() not in seen]
            seen.update(r.idempotency_key() for r in deduped)
            if bad:
                await quarantine(bad)
            if deduped:
                written = await commit(deduped)
                logger.info("committed=%d quarantined=%d", written, len(bad))

    tasks = [asyncio.create_task(handle(batch)) async for batch in source]
    await asyncio.gather(*tasks)

The ObjectRecord model is the enforcement point: strict=True blocks silent type coercion, extra="forbid" rejects payloads carrying unmapped columns from a drifted export, and the rights_uri pattern guarantees no record advances without a machine-readable rights assertion. The idempotency_key method is what upgrades at-least-once broker delivery into effectively-once writes — a retried payload produces the same key and is skipped by the seen set before it ever reaches the database. This coordinator is intentionally thin; the substantive engineering lives in the source generators and commit sinks, which the following workflows specify in depth.

Workflows in This Section #

Each ingress format and orchestration concern has its own dedicated workflow. Together they implement the four gates in the architecture diagram.

Asynchronous ingestion foundations. The building async ingestion pipelines workflow establishes the producer-consumer topology, connection pooling, and semaphore-bounded fetch layer that the coordinator above assumes. It covers the coordinator-yields-batches pattern, httpx connection reuse, and how to keep memory flat during multi-hour digitization runs. Its child guides go deeper on configuring Celery for museum data sync for distributed task orchestration and polling museum APIs with Python requests for cursor-based incremental pulls from external aggregators.

Tabular synchronization. The CSV to database sync strategies workflow handles the most common real-world input: inconsistent CSV dumps from legacy platforms. It specifies chunked async streaming with aiocsv, deterministic primary-key conflict resolution, and ON CONFLICT bulk upserts against PostgreSQL that keep transaction logs from bloating. For campaigns that exceed host memory, handling large CSV batches without memory leaks covers streaming boundaries and garbage-collection tuning.

Unstructured document capture. When the source is a scanned finding aid or a conservation report rather than a structured export, automating OCR metadata extraction inserts a deterministic enrichment stage that turns rasterized documents into structured fields, routes low-confidence extractions to human review, and never blocks batch throughput. Its child guide, extracting provenance text with Tesseract OCR, tunes PSM/OEM parameters for archival card layouts.

The validation gate itself. Schema validation with Pydantic is the deepest treatment of the enforcement layer: strict v2 models, legacy-column aliasing, custom domain validators, and structured error formatting that feeds the quarantine table. Every other workflow in this section routes its output through the model this page describes.

Structured XML and harvest sources. Not every export is tabular. XML & OAI-PMH harvest adapters covers the born-structured ingress path — streaming multi-gigabyte LIDO dumps with lxml iterparse and pulling incremental updates from partner repositories over OAI-PMH — feeding the same Pydantic gate the tabular adapter uses.

Operating the failure branch. Every adapter above routes malformed payloads to a dead-letter queue, and managing the ingestion quarantine queue is where that branch becomes an operational discipline: triaging failures by error class, reconciliation reporting, and safely replaying records once the underlying defect is fixed.

Integration Boundaries #

This pipeline stage does not publish anything — it produces validated, deduplicated records and hands them onward. Understanding the boundaries prevents responsibility bleed between stages. Upstream, the boundary is the message broker: everything before it (capture, export, file drops) is source-system territory, and the ingestion pipeline treats those inputs as untrusted until they pass the validation gate. Downstream, there are two clean handoffs.

The first handoff is to the taxonomy and persistence layer. Once a record validates, normalization resolves its free-text materials, subjects, and geographic terms against Getty authorities and writes both the resolved URI and the human-readable label, following the field-level rules in mapping LIDO to internal databases. The committed shape of the record — indexed columns plus an archival LIDO blob — is defined by designing museum object schemas, not by this pipeline.

The second handoff is to rights and delivery. A committed record carries a validated RightsStatements.org URI, but the policy it implies — whether an asset is embargoed, watermarked, or routed to a public replica — is decided in rights metadata mapping and licensing automation. Only after that policy resolves does a record become eligible for IIIF manifest generation and delivery through a public endpoint. Keeping ingestion ignorant of publication policy is what lets a bulk re-ingest run safely without accidentally re-exposing embargoed material.

Operational Checklist #

Before promoting an ingestion pipeline to production, confirm every gate below. These are the checks that separate a resilient institutional pipeline from a fragile one-off script.

• Message broker enforces at-least-once delivery with visibility timeouts longer than the worst-case batch processing time.
• Every record model carries a stable idempotency_key derived from accession number or manifest URI, and the commit path deduplicates on it.
• Pydantic models run with strict=True and extra="forbid" so column drift in a source export fails loudly instead of writing partial records.
• The rights_uri field is required and pattern-validated; no record can commit without a machine-readable rights assertion.
• Malformed payloads route to a dead-letter queue with the original payload, structured error list, and a UTC quarantine timestamp — never dropped.
• Worker concurrency is bounded by a semaphore sized to database connection-pool limits, not left unbounded.
• Structured logs emit a correlation ID per record traceable across validation, normalization, and commit.
• Metrics export queue depth, per-batch commit/quarantine counts, and validation-failure rate for autoscaling and alerting.
• Bulk upserts use ON CONFLICT (or equivalent) so retries are idempotent at the database boundary.
• A dry-run mode validates a full batch and reports quarantine reasons without writing, for pre-flight verification of a new export.

Failure Modes #

The failures below recur across institutions. Each has a deterministic remediation and a workflow that treats it in depth.

Related #

• Building Async Ingestion Pipelines — bounded-concurrency worker topology
• CSV to Database Sync Strategies — idempotent tabular upserts
• Automating OCR Metadata Extraction — structured fields from scans
• Schema Validation with Pydantic — strict v2 enforcement gate
• XML & OAI-PMH Harvest Adapters — structured and harvested ingress
• Managing the Ingestion Quarantine Queue — dead-letter triage and replay
• Rights Metadata Mapping & Licensing Automation — downstream publication policy