XML & OAI-PMH Harvest Adapters for Museum Records

Within the broader automated record ingestion and sync workflows pipeline, this stage is the structured-XML adapter: it turns namespaced heritage documents — a LIDO harvest from a regional aggregator, an EAD finding aid, or an OAI-PMH feed exposed by another institution’s repository — into validated, deduplicated records inside the collection management system. Where the CSV to database sync strategies adapter owns the messy tabular reality of spreadsheet dumps, and the async pipeline owns born-JSON API payloads, this adapter owns the two problems unique to XML interchange: documents too large to hold in memory, and element names that are meaningless without their namespace.

The failure this adapter prevents is subtle. A naive etree.parse() reads the entire multi-hundred-megabyte LIDO export into a DOM before a single record is emitted, and a naive element.find("lido") returns None on a correctly namespaced document because the tag is really {http://www.lido-schema.org}lido. A production harvest instead streams one record element at a time, clears each element after it is consumed so memory stays flat, resolves every path through an explicit namespace map, and gates each extracted record through a strict Pydantic v2 model before anything touches the database.

Workflow Context #

The three ingestion adapters in this section share one commit-and-quarantine contract but differ entirely in their front door. Choosing the right one is a function of how the source publishes its records.

CSV arrives as a flat file with no schema, so the tabular adapter spends its effort on encoding detection, delimiter sniffing, and coercing string cells into typed fields. A polled JSON API arrives already structured and self-describing, so the async adapter spends its effort on cursor management, connection pooling, and rate limits. XML and OAI-PMH sit between the two: the payload is richly structured and self-validating against a schema, but it is verbose, deeply nested, and namespace-qualified, and an OAI-PMH feed paginates across dozens of HTTP requests stitched together by an opaque resumptionToken. The engineering effort here goes into streaming without exhausting memory and into resolving qualified names correctly — not into guessing types.

Two source shapes flow through this one adapter. A static XML document — a single large LIDO or EAD file dropped on disk by a partner — is consumed with streaming iterparse. A live OAI-PMH endpoint — the harvesting protocol most European aggregators and many university repositories expose — is consumed with a ListRecords request loop that follows resumption tokens until the feed is exhausted. Both converge on the same namespace-qualified extraction and the same validation gate, so a record harvested over OAI-PMH and a record parsed from a file drop reach the commit stage in identical shape. The detailed mechanics of each front door live in parsing LIDO XML with lxml iterparse and harvesting OAI-PMH feeds in Python.

Prerequisites #

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

• Python 3.9+ for the PEP 604 | union syntax and structural pattern matching used in the record dispatcher below.
• lxml (5+) for etree.iterparse, which streams element-by-element and exposes the getprevious/getparent handles needed to release consumed subtrees. The stdlib xml.etree.ElementTree also offers iterparse but lacks the sibling-cleanup primitives that keep memory truly flat on a large document.
• httpx (0.27+) (or requests 2.31+) for the OAI-PMH transport. httpx is preferred for its explicit timeouts and reusable Client connection pool across the resumption-token loop.
• pydantic (v2) for the record contract; this page uses the v2 API (model_config, field_validator, model_validate) — see the pydantic v2 documentation for the strict-mode options.
• The source namespace URIs, declared once as a map: lido → http://www.lido-schema.org for object records, and oai → http://www.openarchives.org/OAI/2.0/ for the OAI-PMH envelope. Every XPath in this adapter resolves through that map — see the LIDO schema and the OAI-PMH specification for the canonical element names.
• A dead-letter store — the same quarantine table the other adapters write to, accepting the raw record XML, the structured error list, and a UTC timestamp.

Schema & Spec Reference #

Two specifications govern this adapter. LIDO defines the content — the element paths from which each field is extracted — and OAI-PMH defines the transport — the verbs and parameters that page through a remote feed. The extraction layer must speak both, because an OAI-PMH ListRecords response wraps LIDO records inside an OAI envelope, so a single field lookup crosses two namespaces.

The LIDO paths below are the minimum set this adapter reads; all are relative to a single lido:lido record element and expressed against the lido namespace prefix.

OAI-PMH is a small verb-oriented HTTP protocol. This adapter uses only the harvesting verbs; the parameters below are what the resumption loop manipulates.

The single most common OAI-PMH implementation bug follows from that last row: on the second and subsequent requests you must send verb and resumptionToken only. Re-sending metadataPrefix or set alongside a token is a protocol error that compliant repositories reject with badArgument.

Step-by-Step Implementation #

1. Declare the namespace map once #

Every qualified lookup in this adapter routes through one shared map. Declaring it as a module constant keeps XPaths readable and guarantees the OAI envelope and the LIDO payload are never confused.

LIDO_NS = "http://www.lido-schema.org"
OAI_NS = "http://www.openarchives.org/OAI/2.0/"

# One map, passed to every find/findtext/iterfind call.
NSMAP: dict[str, str] = {"lido": LIDO_NS, "oai": OAI_NS}

# The Clark-notation qualified tag iterparse filters on.
LIDO_RECORD_TAG = f"{{{LIDO_NS}}}lido"

Edge cases. For CSV there is no namespace concern at all — columns are bare strings. For an API returning JSON there is likewise no namespace, only key paths. XML is the only source where a bare find("lido") silently fails: lxml stores the namespace in the tag itself, so you either pass namespaces=NSMAP with a prefix or match the full {uri}local Clark-notation tag. EAD documents use a different namespace (urn:isbn:1-931666-22-9); add it to the same map rather than hard-coding a second one.

2. Stream lido:lido records with iterparse and clear each element #

iterparse emits an event as each record element closes, so the document is never fully resident. The critical detail is releasing consumed subtrees: after yielding an element, elem.clear() drops its children, and deleting preceding siblings frees the memory the parser would otherwise retain up the tree.

from collections.abc import Iterator
from lxml import etree

def stream_lido_records(source: str) -> Iterator[etree._Element]:
    """Yield each lido:lido element from a large XML file, keeping memory flat."""
    context = etree.iterparse(source, events=("end",), tag=LIDO_RECORD_TAG)
    for _event, elem in context:
        yield elem
        # Release the record's own subtree once the consumer is done with it.
        elem.clear()
        # Drop already-parsed siblings so the root does not accumulate them.
        while elem.getprevious() is not None:
            del elem.getparent()[0]

Edge cases. Yield before clearing — clearing first hands the consumer an empty element. The getprevious loop matters only for the streaming XML path; the CSV adapter bounds memory with a fixed chunk size instead, and the API adapter bounds it with a cursor page size. On a namespaced document, always pass the fully qualified tag= to iterparse; filtering on the bare local name "lido" matches nothing. The full memory-profiling treatment — including why clear() alone is insufficient — lives in parsing LIDO XML with lxml iterparse.

3. Harvest an OAI-PMH feed with ListRecords and resumption tokens #

A live feed paginates. The loop below issues one ListRecords request, yields every oai:record on the page, then follows the resumptionToken until the repository stops returning one. Note the parameter reset on the second iteration.

import httpx
from collections.abc import Iterator
from lxml import etree

def harvest_oai(
    base_url: str,
    metadata_prefix: str = "lido",
    set_spec: str | None = None,
) -> Iterator[etree._Element]:
    """Yield each OAI-PMH record element, following resumption tokens to the end."""
    params: dict[str, str] = {"verb": "ListRecords", "metadataPrefix": metadata_prefix}
    if set_spec:
        params["set"] = set_spec
    with httpx.Client(timeout=30.0) as client:
        while True:
            resp = client.get(base_url, params=params)
            resp.raise_for_status()
            tree = etree.fromstring(resp.content)
            yield from tree.iterfind(".//oai:record", NSMAP)
            token = tree.findtext(".//oai:resumptionToken", namespaces=NSMAP)
            if not token or not token.strip():
                break
            # Once a token exists, verb + token are the ONLY legal arguments.
            params = {"verb": "ListRecords", "resumptionToken": token.strip()}

Edge cases. An empty <resumptionToken/> element on the final page is the protocol’s way of signalling the last batch — treat blank text as end-of-feed, not as a token. Watch for an OAI <error code="..."> element in the envelope: a noRecordsMatch on a from/until window is benign, but badResumptionToken means the token expired mid-harvest and the run must restart. For a CSV or API source this whole stage is absent; only OAI-PMH carries stateful server-side pagination. The retry, backoff, and token-expiry handling are detailed in harvesting OAI-PMH feeds in Python.

4. Extract namespace-qualified fields and validate into Pydantic #

Both front doors converge here. Given a lido:lido element — whether streamed from a file or unwrapped from an OAI envelope — the extractor pulls each field through the namespace map into a plain dict, then the strict model gates it. Conforming records become instances; failures carry their structured error list to quarantine without aborting the harvest.

from datetime import date
from lxml import etree
from pydantic import BaseModel, ConfigDict, ValidationError, field_validator

RIGHTS_URI_PREFIX = "https://rightsstatements.org/vocab/"

class HarvestedRecord(BaseModel):
    """Strict contract for one harvested LIDO record destined for the CMS."""
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    lido_rec_id: str
    title: str
    creator: str | None = None
    rights_uri: str

    @field_validator("lido_rec_id", "title")
    @classmethod
    def not_blank(cls, value: str) -> str:
        # A blank record identifier would corrupt the idempotency key.
        if not value:
            raise ValueError("required field is blank")
        return value

    @field_validator("rights_uri")
    @classmethod
    def must_be_rights_uri(cls, value: str) -> str:
        if not value.startswith(RIGHTS_URI_PREFIX):
            raise ValueError(f"rights_uri must be a {RIGHTS_URI_PREFIX} URI")
        return value


def extract_fields(elem: etree._Element) -> dict[str, str | None]:
    """Pull a flat field dict from one namespace-qualified LIDO element."""
    def text(xpath: str) -> str | None:
        found = elem.findtext(xpath, namespaces=NSMAP)
        return found.strip() if found else None

    return {
        "lido_rec_id": text("lido:lidoRecID"),
        "title": text(".//lido:titleSet/lido:appellationValue"),
        "creator": text(
            ".//lido:actorInRole/lido:actor/lido:nameActorSet/lido:appellationValue"
        ),
        "rights_uri": text(".//lido:rightsResource/lido:rightsType/lido:conceptID"),
    }


def validate_records(
    elements: Iterator[etree._Element],
) -> tuple[list[HarvestedRecord], list[dict[str, object]]]:
    """Split a stream of LIDO elements into (valid records, quarantined rows)."""
    valid: list[HarvestedRecord] = []
    rejected: list[dict[str, object]] = []
    for elem in elements:
        raw = extract_fields(elem)
        try:
            valid.append(HarvestedRecord.model_validate(raw))
        except ValidationError as exc:
            rejected.append({
                "raw": {k: v for k, v in raw.items()},
                "errors": exc.errors(include_url=False),
            })
    return valid, rejected

Edge cases. findtext returns None for a missing path, which the strict model then rejects for a required field — exactly the desired behaviour, since a LIDO record with no lidoRecID cannot be keyed. A field with cardinality 1..n (multiple titles) returns only the first with findtext; use iterfind and join, or keep the primary and route the rest to a repeatable table. For a lenient first-pass migration, relax the model with str | None defaults; for the authoritative production harvest keep extra="forbid" strict, matching the contract the schema validation with Pydantic guide specifies for every adapter in this section.

Rights and Access Routing #

Rights metadata decides visibility, not just descriptive completeness — and in a LIDO harvest the rights assertion arrives as a lido:conceptID inside a lido:rightsResource block, which is why the extractor pulls it as a first-class field rather than treating it as optional prose. Once must_be_rights_uri confirms the value is a resolvable RightsStatements.org URI, that URI determines the access tier the record receives at commit time. A record carrying InC/1.0/ (in copyright) commits at a metadata-only tier — the descriptive record is public but the media asset is gated — while NoC-US/1.0/ and NKC/1.0/ records may expose full-resolution media downstream.

Harvested records carry a specific hazard here: a partner repository may assert rights that conflict with your own institution’s determination for the same object, or may leave the rightsResource block empty. An empty or non-URI rights value must never default to “public.” Route it to the quarantine queue exactly as a schema failure would be routed, and let a curator reconcile it before the record becomes resolvable. Embargo handling — records that stay dark until a future release date — is a separate routing dimension covered in implementing embargo workflows, and the resolved URI is later written into the IIIF manifest rights property so every downstream viewer enforces the same policy the harvest asserted, per the IIIF Presentation API 3.0 specification.

Verification and Testing #

Before pointing the harvest at a production database, prove extraction and validation against a small inline document. The assert-based test below exercises namespace-qualified extraction, a valid record, and a blank-identifier rejection in isolation — no network or file required.

from lxml import etree

SAMPLE_LIDO = f"""
<lido:lido xmlns:lido="{LIDO_NS}">
  <lido:lidoRecID>OBJ-000123</lido:lidoRecID>
  <lido:descriptiveMetadata>
    <lido:objectIdentificationWrap>
      <lido:titleWrap><lido:titleSet>
        <lido:appellationValue>Portrait of a Collector</lido:appellationValue>
      </lido:titleSet></lido:titleWrap>
    </lido:objectIdentificationWrap>
  </lido:descriptiveMetadata>
  <lido:administrativeMetadata>
    <lido:rightsWorkWrap><lido:rightsWorkSet><lido:rightsResource>
      <lido:rightsType>
        <lido:conceptID>https://rightsstatements.org/vocab/InC/1.0/</lido:conceptID>
      </lido:rightsType>
    </lido:rightsResource></lido:rightsWorkSet></lido:rightsWorkWrap>
  </lido:administrativeMetadata>
</lido:lido>
"""

def test_extract_and_validate() -> None:
    elem = etree.fromstring(SAMPLE_LIDO.encode("utf-8"))

    # 1. Namespace-qualified extraction resolves through the shared map.
    raw = extract_fields(elem)
    assert raw["lido_rec_id"] == "OBJ-000123"
    assert raw["title"] == "Portrait of a Collector"

    # 2. A well-formed record validates.
    record = HarvestedRecord.model_validate(raw)
    assert record.rights_uri.startswith(RIGHTS_URI_PREFIX)

    # 3. A blank identifier is rejected, not committed.
    raw_bad = {**raw, "lido_rec_id": ""}
    try:
        HarvestedRecord.model_validate(raw_bad)
        assert False, "blank lido_rec_id should have raised"
    except ValidationError:
        pass

if __name__ == "__main__":
    test_extract_and_validate()
    print("harvest contract OK")

Then run a full dry run that streams and validates the real source but skips the commit, so you can count survivors and quarantined records before writing anything:

python -m harvest.dryrun --oai https://partner.example.org/oai --prefix lido --no-commit

A clean run reports zero unexpected rejections and a survivor count matching the feed’s advertised record total minus known-bad records. Any surprise in that delta is namespace drift or a schema mismatch to investigate before the production harvest.

Where to Go Deeper #

The two front doors each carry enough operational depth to warrant a dedicated guide.

Parsing LIDO XML with lxml iterparse profiles the streaming path in detail: why elem.clear() alone leaves memory climbing, how the getprevious sibling-deletion loop releases the retained tree, how to filter iterparse on a Clark-notation tag, and how to fold the same technique onto EAD finding aids that nest records differently from LIDO.

Harvesting OAI-PMH Feeds in Python covers the transport path: the full resumption-token loop with retry and backoff, incremental from/until datestamp windows for delta syncs, distinguishing benign noRecordsMatch from a fatal badResumptionToken, and mapping the harvested LIDO into your internal schema following mapping LIDO to internal databases.

FAQ #

Why does element.find("lido:lidoRecID") return None on a valid LIDO file? #

Because lxml stores the namespace inside the tag, so a bare prefix without a namespace map resolves to nothing. Pass namespaces=NSMAP (mapping lido to http://www.lido-schema.org) on every find, findtext, and iterfind call, or match the fully qualified {http://www.lido-schema.org}lidoRecID Clark-notation tag directly. The element is present; the query simply is not asking for the right qualified name.

My harvester runs out of memory on a large XML export. What am I doing wrong? #

You are almost certainly using etree.parse(), which builds the entire DOM before returning. Switch to etree.iterparse(source, events=("end",), tag=LIDO_RECORD_TAG) and, after yielding each record, call elem.clear() and delete preceding siblings with the while elem.getprevious() is not None: del elem.getparent()[0] loop. Without the sibling deletion, the root element silently retains every parsed record and memory still climbs.

What is a resumption token and why does my second OAI-PMH request fail? #

A resumptionToken is the opaque, stateful cursor an OAI-PMH repository returns to page through a large result set. The failure is almost always that you re-sent metadataPrefix or set alongside the token. Once a token is issued, the only legal arguments are verb and resumptionToken; sending anything else is a badArgument protocol error. Reset your parameter dict to exactly those two keys on every follow-up request.

How do I harvest only records changed since the last run? #

Use an incremental window: add from and until parameters in ISO 8601 UTC to the initial ListRecords request, then follow resumption tokens as usual. Persist the harvest’s completion timestamp and pass it as the next run’s from value. A window that matches nothing returns an oai:error with code="noRecordsMatch", which is benign and should be treated as “zero new records,” not as a failure.

Should XML records and CSV rows share the same validation model? #

They should share the same contract, not necessarily the same model class. Both adapters must enforce the identical field requirements, rights-URI check, and extra="forbid" strictness so a record’s provenance never changes what “valid” means. In practice you extract each source into a common flat dict, then validate against one shared model — which is exactly how this adapter and the CSV adapter both converge on the commit-and-quarantine guarantees.

Related #

• Automated Record Ingestion & Sync Workflows — parent pipeline overview
• Parsing LIDO XML with lxml iterparse — flat-memory streaming deep dive
• Harvesting OAI-PMH Feeds in Python — resumption-token transport guide
• CSV to Database Sync Strategies — sibling flat-file adapter
• Schema Validation with Pydantic — the shared strict v2 contract
• Mapping LIDO to Internal Databases — persisting the harvested shape