Building Async Ingestion Pipelines for Museum Digital Assets

Within the broader automated record ingestion and sync workflows pipeline, this stage is the concurrent fetch-and-validate engine that turns a queue of object identifiers into committed, standards-conformant collection records. It owns the transport layer: pulling payloads from a digitization API or object store, bounding how many requests run at once, and handing only schema-clean records to the database. This page specifies the worker topology, the data contract that gates every record, the retry logic that survives museum API maintenance windows, and the verification harness a Python automation engineer runs before a bulk digitization campaign goes live.

The problem this solves is narrow but unforgiving. A synchronous script that walks forty thousand accession records with requests.get() blocks on every round trip, opens and discards a TCP socket per call, and loads the whole result set into a list until the digitization host runs out of memory. An asynchronous design decouples network wait from validation work so hundreds of fetches can be in flight against a fixed connection pool, while a semaphore caps concurrency to what the upstream API and the target database can actually absorb.

Workflow Context #

Museum digital asset ingestion runs under two constraints that generic web scraping never faces: strict metadata conformance and non-negotiable rights handling. Every record must arrive typed and validated against institutional standards before it touches the collection management system, and no record may become publicly resolvable until its rights status is asserted. This workflow covers the concurrent path for born-structured sources — API payloads and object-storage events. Born-unstructured scans take a different route first through automating OCR metadata extraction, and flat tabular exports flow through CSV to database sync strategies; all three converge on the same validation and commit gates described here.

The design follows a strict producer-consumer topology with three independently bounded stages. A coordinator yields batches of object identifiers from a staging queue. A pool of async workers fetches payloads concurrently through a shared connection pool, throttled by a semaphore. A validation gate enforces the record contract and splits the stream — conforming records proceed to a bulk upsert, malformed payloads divert to a dead-letter queue for forensic review. Separating fetch, validate, and commit prevents a slow database from starving the network layer and stops a burst of bad payloads from halting the whole run.

Pipeline Architecture & Data Flow #

The staging queue holds only identifiers, not payloads, so restarting a failed run costs nothing but re-fetching. The coordinator streams batches rather than materializing the full ID set, which keeps memory flat regardless of campaign size. Each worker acquires a semaphore slot before issuing a request and releases it on completion, so the number of open sockets never exceeds the pool the upstream API tolerates. Validation runs off the network path, and the dead-letter queue preserves every rejected payload with its structured error list so nothing is silently dropped.

Every arrow is an explicit gate, not a best-effort handoff. The semaphore gate absorbs concurrency pressure; the validation gate enforces the metadata contract; the upsert gate enforces idempotency at the database boundary so a retried batch never writes a record twice.

Prerequisites #

Before wiring the pipeline, confirm the following are in place:

• Python 3.9+ for the PEP 604 | union syntax and asyncio improvements used throughout.
• httpx (0.27+) as the async HTTP client — its AsyncClient supports connection-pool limits and per-request timeouts that requests cannot express.
• pydantic (v2) for the record contract; this page uses the v2 API (model_config, field_validator, model_dump) — see the pydantic v2 documentation for migration notes.
• A staging queue or cursor source — a database table of pending object_ids, an SQS/Redis queue, or an API pagination cursor. Legacy polling cursors are covered in polling museum APIs with Python requests.
• Target CMS write access exposing a bulk upsert with an ON CONFLICT (or equivalent) clause keyed on accession number.
• Standards endpoints reachable at validation time: the IIIF Presentation API 3.0 specification for manifest inputs and the LIDO schema as the interchange envelope. Free-text materials and place names resolve against Getty authorities — see implementing Getty AAT & TGN.
• A dead-letter store — a quarantine table or bucket that accepts the raw payload, the error list, and a UTC timestamp.

Data Contract & Schema Reference #

The record contract is the single point of truth for what a valid asset looks like. Raw API responses rarely match institutional metadata standards, so a strict Pydantic v2 model coerces types, enforces field presence, and rejects unknown keys before any downstream work. The model below mirrors the fields the collection management system requires and encodes the controlled-vocabulary constraints that free-text sources routinely violate.

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

RIGHTS_PATTERN = r"^https://rightsstatements\.org/vocab/[A-Za-z\-]+/1\.0/$"

class AssetRecord(BaseModel):
    # Strict mode + extra="forbid" makes source column drift fail loudly
    # instead of silently writing a partial record.
    model_config = ConfigDict(strict=True, extra="forbid")

    accession_number: str
    title: str = Field(min_length=1)
    media_url: str
    rights_uri: str = Field(pattern=RIGHTS_PATTERN)
    embargo_date: date | None = None
    access_tier: str = Field(default="public")
    iiif_manifest: str | None = None
    lido_namespace: str = Field(default="http://www.lido-schema.org/")

    @field_validator("access_tier")
    @classmethod
    def validate_tier(cls, v: str) -> str:
        allowed = {"public", "research", "restricted"}
        if v not in allowed:
            raise ValueError(f"access_tier must be one of {sorted(allowed)}")
        return v

Because extra="forbid" is set, a legacy export that adds an unmapped column fails validation instead of quietly discarding it — the failure surfaces at the dead-letter queue where it can be reconciled. The deeper contract-design patterns, including field aliases for legacy TMS column names, live in schema validation with Pydantic.

Step-by-Step Implementation #

1. Bound concurrency with a shared client and semaphore #

Unbounded concurrency triggers upstream rate limits and exhausts database connection slots. The pipeline holds one AsyncClient for the whole run — reusing its connection pool across every request — and gates each fetch through an asyncio.Semaphore sized to what the API tolerates.

import asyncio
import logging
from collections.abc import AsyncIterator
from typing import Any
import httpx

logger = logging.getLogger(__name__)

class AsyncIngestionPipeline:
    def __init__(self, api_base: str, max_concurrency: int = 15):
        self.api_base = api_base
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.client = httpx.AsyncClient(
            timeout=10.0,
            limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
        )

    async def __aenter__(self) -> "AsyncIngestionPipeline":
        return self

    async def __aexit__(self, *exc_info: Any) -> None:
        await self.client.aclose()

Size max_concurrency to the smaller of the API’s rate ceiling and the database connection pool — if the CMS pool holds twenty connections, running fifty concurrent commits only produces QueuePool limit errors downstream.

2. Fetch with retry and exponential backoff #

Museum APIs return 429 and 503 during maintenance windows and under digitization-lab load. Transient failures retry with exponential backoff plus jitter to avoid a thundering-herd retry storm; permanent 4xx errors (other than 429) raise immediately so they route to quarantine rather than wasting retry budget.

import random

TRANSIENT = {429, 500, 502, 503, 504}

async def fetch_asset(self, asset_id: str, max_retries: int = 4) -> dict[str, Any]:
    url = f"{self.api_base}/assets/{asset_id}"
    for attempt in range(max_retries):
        async with self.semaphore:  # hold a slot only for the actual request
            try:
                resp = await self.client.get(url)
                resp.raise_for_status()
                return resp.json()
            except httpx.HTTPStatusError as exc:
                status = exc.response.status_code
                if status not in TRANSIENT or attempt == max_retries - 1:
                    raise  # permanent, or out of retries -> caller quarantines
            except (httpx.TimeoutException, httpx.TransportError):
                if attempt == max_retries - 1:
                    raise
        # backoff runs OUTSIDE the semaphore so a sleeping worker frees its slot
        wait = min(2 ** attempt + random.uniform(0, 1), 30.0)
        logger.warning("Retry %d for %s in %.1fs", attempt + 1, asset_id, wait)
        await asyncio.sleep(wait)
    raise RuntimeError(f"unreachable: {asset_id}")  # pragma: no cover

The critical detail is that the backoff sleep happens after releasing the semaphore — a worker that is waiting to retry must not hold a concurrency slot, or a wave of 429s will deadlock the pool.

3. Validate a batch and split the stream #

asyncio.gather(..., return_exceptions=True) lets one failed fetch fail without cancelling its siblings. Each successful payload is validated; a ValidationError diverts the record to the dead-letter queue with its structured error list instead of aborting the batch.

from datetime import datetime, timezone

async def process_batch(self, batch_ids: list[str]) -> AsyncIterator[AssetRecord]:
    tasks = [self.fetch_asset(_id) for _id in batch_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for asset_id, result in zip(batch_ids, results):
        if isinstance(result, Exception):
            await self.quarantine(asset_id, {"fetch_error": str(result)})
            continue
        try:
            yield AssetRecord(**result)
        except ValidationError as exc:
            await self.quarantine(asset_id, {
                "raw": result,
                "errors": exc.errors(),
                "quarantined_at": datetime.now(timezone.utc).isoformat(),
            })

async def quarantine(self, asset_id: str, detail: dict[str, Any]) -> None:
    logger.error("Quarantined %s: %s", asset_id, detail.get("errors", detail))
    # persist `detail` to the dead-letter table/bucket keyed by asset_id

Note the edge-case handling per source variant: an API payload arrives as a dict and unpacks directly into the model; an XML export must be parsed to a dict first (deferring namespace resolution to the transform stage); a CSV row needs extra="forbid" to catch drifted columns before they reach the model. All three converge on the same AssetRecord contract.

4. Coordinate batches and commit idempotently #

The coordinator streams identifiers in fixed-size batches and accumulates only the small window of validated records for one bulk upsert, keeping memory flat. The upsert uses ON CONFLICT (accession_number) so a redelivered or retried batch updates in place rather than creating duplicates.

async def run(self, id_source: AsyncIterator[str], batch_size: int = 200) -> None:
    batch: list[str] = []
    async for asset_id in id_source:
        batch.append(asset_id)
        if len(batch) >= batch_size:
            await self._flush(batch)
            batch = []
    if batch:
        await self._flush(batch)

async def _flush(self, batch_ids: list[str]) -> None:
    records = [r async for r in self.process_batch(batch_ids)]
    if records:
        rows = [r.model_dump(mode="json") for r in records]
        await self.bulk_upsert(rows)  # INSERT ... ON CONFLICT (accession_number) DO UPDATE
        logger.info("Committed %d of %d", len(records), len(batch_ids))

Streaming the ID source with async for rather than reading it into a list is what lets the same coordinator process two hundred records or two hundred thousand without the memory profile changing.

Rights and Access Routing #

No record becomes publicly resolvable until its rights status is asserted, so the pipeline treats rights as a gate, not a decoration. The rights_uri field is required and pattern-matched against the RightsStatements.org vocabulary, so a payload missing a machine-readable rights statement fails validation and quarantines rather than committing as public by default. The embargo_date and access_tier fields then determine visibility: a record with a future embargo_date is forced to a non-public tier at commit time regardless of what the source claimed.

def resolve_access(record: AssetRecord, today: date) -> str:
    # A future embargo always overrides an optimistic source tier.
    if record.embargo_date and record.embargo_date > today:
        return "restricted"
    return record.access_tier

This keeps the transport layer honest about intellectual-property state while deferring licensing policy — Creative Commons routing, orphan-work handling, embargo lifecycle — to rights metadata mapping and licensing automation, which consumes the tier this stage assigns.

Verification and Testing #

Before a campaign runs against production, exercise the validation gate and the access-routing logic with a fast, dependency-free test. This asserts that a clean payload validates, a drifted column is rejected, and a future embargo is downgraded.

import pytest
from datetime import date

def test_valid_record_passes():
    rec = AssetRecord(
        accession_number="1998.42.1",
        title="Portrait study",
        media_url="https://assets.example.org/1998.42.1.jpg",
        rights_uri="https://rightsstatements.org/vocab/InC/1.0/",
    )
    assert rec.access_tier == "public"

def test_unknown_column_is_rejected():
    with pytest.raises(ValidationError):
        AssetRecord(
            accession_number="1998.42.2",
            title="Untitled",
            media_url="https://assets.example.org/x.jpg",
            rights_uri="https://rightsstatements.org/vocab/CNE/1.0/",
            legacy_dept_code="PAINT",  # not in the contract -> fails on extra="forbid"
        )

def test_future_embargo_downgrades_tier():
    rec = AssetRecord(
        accession_number="1998.42.3",
        title="Sealed bequest",
        media_url="https://assets.example.org/y.jpg",
        rights_uri="https://rightsstatements.org/vocab/InC/1.0/",
        embargo_date=date(2099, 1, 1),
        access_tier="public",
    )
    assert resolve_access(rec, date(2026, 7, 3)) == "restricted"

Run it with pytest -q; all three cases should pass. For a live dry run, add a mode that calls process_batch against a sample of real identifiers and reports quarantine reasons without calling bulk_upsert — a pre-flight check that surfaces a drifted export before it corrupts the catalogue.

In This Section #

Two focused guides go deeper on the transport concerns this architecture depends on:

• Polling Museum APIs with Python Requests — the synchronous-cursor variant: resolving 429/503 rate-limit exhaustion, ephemeral-port starvation, and pagination state drift when a full async pool is more than a sync mirror job needs.
• Configuring Celery for Museum Data Sync — pushing bounded ingestion onto a distributed task queue: broker pool limits, worker_prefetch_multiplier, and taming worker memory growth on high-volume overnight syncs.

FAQ #

Why does my worker pool deadlock during a burst of 429 responses? #

Almost always because the backoff sleep runs while the worker still holds its semaphore slot. When every worker is sleeping-with-a-slot, no new request can start and the pool stalls. Release the semaphore before sleeping — as in the fetch loop above, the async with self.semaphore block wraps only the request, and asyncio.sleep(wait) sits outside it.

How do I size max_concurrency correctly? #

Set it to the smaller of the upstream API’s documented rate ceiling and your database connection-pool size. Concurrency higher than the CMS pool does not speed commits; it just converts fetch throughput into QueuePool limit reached errors at the write boundary. Start conservative (10–15), watch queue depth and the API’s X-RateLimit-Remaining header, and raise it only if both have headroom.

A few bad records abort my whole batch. How do I isolate failures? #

Use asyncio.gather(*tasks, return_exceptions=True). Without return_exceptions=True, the first exception cancels every sibling task in the batch. With it, failures are returned as values you can inspect per record and route individually to the dead-letter queue, so one malformed payload never takes down the other 199.

Why do valid-looking payloads still fail validation? #

With model_config = ConfigDict(strict=True, extra="forbid"), two common causes are an unmapped extra column in the source export (rejected by extra="forbid") and a type that would otherwise be silently coerced — for example an integer accession_number where the contract expects a string. This is intentional: strict mode surfaces source drift at the quarantine gate instead of writing partial records. Inspect exc.errors() in the quarantine payload to see the exact field and rule.

Memory climbs steadily during a large campaign until the host is OOM-killed. #

You are accumulating instead of streaming. Reading the identifier source into a list, or collecting every validated record before a single commit, makes memory scale with campaign size. Stream the ID source with async for, flush in fixed-size batches, and let each batch’s records go out of scope after its upsert — the memory profile then stays flat whether you ingest two hundred records or two hundred thousand.

Related #

• Automated Record Ingestion & Sync Workflows — parent pipeline overview
• Polling Museum APIs with Python Requests — synchronous cursor variant
• Configuring Celery for Museum Data Sync — distributed task-queue orchestration
• Celery vs Prefect for Museum Batch Jobs — orchestration tooling decision
• Schema Validation with Pydantic — strict v2 contract design
• Rights Metadata Mapping & Licensing Automation — downstream access policy