Operational Context

A Python automation engineer is asked to keep an aggregation database in step with a partner repository that exposes its records over an OAI-PMH endpoint. The first pass — a single ListRecords request — returns a few hundred records and looks finished, but the partner’s collection holds two hundred thousand. What the first response actually returned was one page, terminated by a resumptionToken, and the harvest was supposed to keep requesting until the token ran out. Worse, the nightly re-run either re-downloads the entire repository every night, or skips records the partner edited after the last run, because there is no persisted cursor telling the job where the previous window ended. And when the partner removes a record, nothing local deletes it: the object lingers in the aggregation long after it vanished upstream, because a deletion in OAI-PMH is not an absent record — it is a present record carrying status="deleted" that the harvester has to notice.

This page resolves that exact task: a correct, incremental OAI-PMH harvest loop that follows resumption tokens to exhaustion, requests only the from/until window that changed since last time, and routes deleted-record tombstones to a purge instead of dropping them on the floor. It is the harvest front door of the XML and OAI-PMH harvest adapters stage, and it feeds the same validated-record contract the rest of the automated record ingestion and sync workflows pipeline depends on.

Root Cause Analysis

Broken OAI-PMH harvests almost always trace to one of three protocol facts that a naive single-request script ignores.

First, pagination is a stateful server cursor, not a page number you control. When a repository cannot return every matching record in one response, it appends a resumptionToken and expects you to send it back — with the verb and nothing else — to retrieve the next page. Two consequences follow. The token is opaque and server-side, so you cannot compute the “next” one or skip ahead; you can only echo what you were handed. And the token expires: repositories bound how long a cursor stays live, so a harvest that pauses too long between pages, or resumes a token persisted from a previous run, gets a badResumptionToken error rather than the next page. A stale token cannot be retried into life — the whole window has to be re-requested from a persisted date cursor.

Second, incremental harvesting needs a durable date cursor, and the window boundaries are subtle. Selective harvesting is expressed with from and until UTCdatetime arguments. To sync only what changed, the lower bound of tonight’s window must be the high-water mark of last night’s successful run, and that watermark has to survive a crash. The upper bound must be frozen at the start of the run, not left open, so that records the partner adds while the harvest is running are caught by the next pass rather than straddling the boundary and being missed. Get either bound wrong and you silently drop edits.

Third, deletions arrive as records, not as absences. A repository that supports deleted-record tracking re-sends a removed object inside the window with a header status="deleted" and no <metadata> element. A harvester that only looks for metadata parses these as empty and ignores them, so the local copy is never removed. Deleted records must be treated as a routable signal — validate the live ones, purge the tombstoned ones — exactly as the async ingestion pipeline treats a dead-letter as a state rather than an error.

Canonical Solution

The harvest loop is a generator: it fetches a page, parses each <record> into a small dataclass that records whether the record is a tombstone, yields them, and follows the resumptionToken until the element comes back empty or absent — the protocol’s signal that the last page was reached. Selective-harvest arguments (metadataPrefix, set, from, until) are sent on the first request only; every continuation carries the verb and the token alone.

python
from __future__ import annotations

import time
from collections.abc import Iterator
from dataclasses import dataclass
from xml.etree import ElementTree as ET

import requests

OAI = "http://www.openarchives.org/OAI/2.0/"


class OAIError(RuntimeError):
    """The server returned an <error> element other than an expired token."""


class TokenExpired(RuntimeError):
    """A resumptionToken was rejected as stale — the window must restart."""


@dataclass(frozen=True)
class HarvestRequest:
    endpoint: str
    metadata_prefix: str = "oai_dc"   # "lido" for full LIDO records; oai_dc is mandatory
    set_spec: str | None = None       # one OAI set, or None for the whole repository
    from_date: str | None = None      # inclusive lower bound, e.g. "2026-07-16T00:00:00Z"
    until_date: str | None = None     # inclusive upper bound, frozen at run start


@dataclass(frozen=True)
class Record:
    identifier: str
    datestamp: str
    deleted: bool
    metadata: ET.Element | None       # None for a deleted-record tombstone


def _q(tag: str) -> str:
    return f"{{{OAI}}}{tag}"           # expand a bare tag into its OAI-PMH namespace


def _first_params(req: HarvestRequest) -> dict[str, str]:
    # Selective-harvest arguments travel ONCE, on the first request. After a token is
    # issued, verb + resumptionToken are the ONLY legal arguments; resending from/until/
    # set/metadataPrefix beside a token is a badArgument protocol error.
    params = {"verb": "ListRecords", "metadataPrefix": req.metadata_prefix}
    if req.set_spec:
        params["set"] = req.set_spec
    if req.from_date:
        params["from"] = req.from_date
    if req.until_date:
        params["until"] = req.until_date
    return params


def _fetch(session: requests.Session, endpoint: str, params: dict[str, str]) -> ET.Element:
    resp = session.get(endpoint, params=params, timeout=30)
    resp.raise_for_status()
    root = ET.fromstring(resp.content)
    err = root.find(_q("error"))
    if err is not None:
        code = err.get("code", "")
        if code == "badResumptionToken":
            # Tokens are stateful server cursors and they expire. A stale token cannot be
            # retried into life — the caller must re-open the window from its saved cursor.
            raise TokenExpired(err.text or "resumptionToken expired")
        if code == "noRecordsMatch":
            return root                # an empty but valid window, not a failure
        raise OAIError(f"{code}: {err.text}")
    return root


def _parse_page(root: ET.Element) -> tuple[list[Record], str | None]:
    listing = root.find(_q("ListRecords"))
    if listing is None:
        return [], None
    records: list[Record] = []
    for rec in listing.findall(_q("record")):
        header = rec.find(_q("header"))
        deleted = header.get("status") == "deleted"       # the tombstone marker
        meta = rec.find(_q("metadata"))
        payload = meta[0] if (meta is not None and len(meta)) else None
        records.append(Record(
            identifier=header.findtext(_q("identifier"), ""),
            datestamp=header.findtext(_q("datestamp"), ""),
            deleted=deleted,
            metadata=None if deleted else payload,        # deleted records carry no metadata
        ))
    token_el = listing.find(_q("resumptionToken"))
    text = token_el.text.strip() if (token_el is not None and token_el.text) else ""
    # An element that is ABSENT or PRESENT-BUT-EMPTY both mean "last page reached".
    return records, (text or None)


def harvest(
    req: HarvestRequest,
    session: requests.Session | None = None,
    throttle: float = 0.0,
) -> Iterator[Record]:
    session = session or requests.Session()
    params = _first_params(req)
    while True:
        root = _fetch(session, req.endpoint, params)
        records, token = _parse_page(root)
        yield from records
        if not token:                                     # empty/absent token → terminate
            return
        params = {"verb": "ListRecords", "resumptionToken": token}
        if throttle:
            time.sleep(throttle)                          # be polite to the partner endpoint

The loop terminates because the only way it continues is a non-empty token, and a conformant repository issues an empty token on the final page. Around that core, incremental syncing is a thin wrapper that persists a single value — the window’s upper bound — and reuses it as next run’s lower bound. Freezing until at run start and writing the cursor only after a clean pass makes the job idempotent: a crash mid-window simply replays the same window.

python
import json
from datetime import datetime, timezone
from pathlib import Path


def run_incremental(base: HarvestRequest, cursor_path: str) -> int:
    # Lower bound = last successful run's high-water mark. Upper bound = now, frozen, so
    # records the partner adds mid-run are caught by the NEXT pass, never straddle the edge.
    cursor = Path(cursor_path)
    from_date = json.loads(cursor.read_text())["until"] if cursor.exists() else None
    until = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    req = HarvestRequest(
        endpoint=base.endpoint,
        metadata_prefix=base.metadata_prefix,
        set_spec=base.set_spec,
        from_date=from_date,          # None on the very first run = a full harvest
        until_date=until,
    )

    seen = 0
    for record in harvest(req):
        seen += 1
        if record.deleted:
            purge(record.identifier)                      # remove the local copy of a tombstone
        else:
            validate_and_upsert(record.identifier, record.metadata)

    # Advance the cursor ONLY after the whole window drained cleanly.
    cursor.write_text(json.dumps({"until": until}))
    return seen

Because harvest yields one record at a time and never accumulates a page, memory stays flat across a repository of any size, matching the streaming discipline in handling large CSV batches without memory leaks.

OAI-PMH harvest loop: token continuation with a deleted-record branch A first ListRecords request with selective from, until, set and metadataPrefix arguments flows into a parse-page step that emits records and a resumptionToken. Each record branches on its header status: deleted records become tombstones that purge the local copy, live records are yielded to validation. A decision then checks whether a resumptionToken is present. If present, a continuation request carrying only the verb and token loops back into parse-page. If empty or absent, the harvest is complete. yes · verb + resumptionToken only First request from · until · set Parse page records + token token present? Harvest complete status = deleted? Tombstone · purge deleted = True Yield to validation live record no per record deleted live

Edge Cases and Variants

  • oai_dc versus lido metadataPrefix. Every OAI-PMH repository must expose oai_dc (flat Dublin Core), which is enough for discovery but lossy — it collapses structured events, actors, and rights into strings. When the partner also advertises a lido prefix, request that instead to keep full structured metadata; each <record>'s <metadata> then wraps a <lido:lido> element you hand straight to parsing LIDO XML with lxml iterparse. Call ListMetadataFormats first to see which prefixes the endpoint actually supports.
  • Token expiry mid-harvest. A repository may expire a live cursor if a page takes too long to process. Catch TokenExpired, and rather than retrying the dead token, restart the harvest from the persisted from cursor — the frozen until bound makes the replayed window deterministic. Never persist a resumption token between runs; persist the date watermark.
  • Full versus incremental. The first ever run has no cursor, so from_date is None and the job harvests the entire repository. Every subsequent run is bounded by the saved watermark. A periodic full re-harvest (no from) is still worth scheduling to catch a repository that under-reports deletions.
  • Set-scoped harvesting. Passing set_spec restricts the window to one OAI set — useful when a partner exposes several departments under one endpoint. Sets narrow which records return; they do not change the token or deletion mechanics.
  • XML versus API versus CSV sources. OAI-PMH is a paginated XML transport; the token loop here is its analogue of the cursor pagination in polling museum APIs with Python requests and the offset checkpoint in a CSV sync. Only the continuation mechanism differs — token, cursor, or byte offset — while the streaming, validation, and idempotency guarantees are identical.

Validation

The invariant that matters most is that token pagination terminates — that the loop stops on the empty final token rather than hanging or re-requesting forever. Prove it, plus deleted-record flagging and the first-request-only argument rule, with a stub session that serves two canned pages and needs no network.

python
# test_oai_harvest.py  ->  run with:  pytest -q test_oai_harvest.py

_PAGE_ONE = b"""<?xml version="1.0" encoding="UTF-8"?>
<OAI-PMH xmlns="http://www.openarchives.org/OAI/2.0/">
  <ListRecords>
    <record>
      <header><identifier>oai:museum:1</identifier><datestamp>2026-07-16</datestamp></header>
      <metadata><oai_dc xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/"/></metadata>
    </record>
    <record>
      <header status="deleted"><identifier>oai:museum:2</identifier><datestamp>2026-07-16</datestamp></header>
    </record>
    <resumptionToken>TOKEN-abc</resumptionToken>
  </ListRecords>
</OAI-PMH>"""

_PAGE_TWO = b"""<?xml version="1.0" encoding="UTF-8"?>
<OAI-PMH xmlns="http://www.openarchives.org/OAI/2.0/">
  <ListRecords>
    <record>
      <header><identifier>oai:museum:3</identifier><datestamp>2026-07-16</datestamp></header>
      <metadata><oai_dc xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/"/></metadata>
    </record>
    <resumptionToken/>
  </ListRecords>
</OAI-PMH>"""


class _Resp:
    def __init__(self, content: bytes) -> None:
        self.content = content

    def raise_for_status(self) -> None:
        pass


class _Session:
    """Serves canned pages and records the arguments each call actually sent."""

    def __init__(self, pages: list[bytes]) -> None:
        self._pages = pages
        self.calls: list[dict] = []

    def get(self, endpoint: str, params: dict, timeout: int) -> _Resp:
        self.calls.append(dict(params))
        return _Resp(self._pages[len(self.calls) - 1])


def test_pagination_terminates_and_flags_deleted() -> None:
    session = _Session([_PAGE_ONE, _PAGE_TWO])
    req = HarvestRequest(endpoint="https://partner.example/oai", from_date="2026-07-15")
    records = list(harvest(req, session=session))

    # The loop stops on the empty resumptionToken — it does not hang on page two.
    assert len(session.calls) == 2
    assert [r.identifier for r in records] == ["oai:museum:1", "oai:museum:2", "oai:museum:3"]

    # The deleted record is a tombstone: flagged, and carrying no metadata.
    deleted = [r for r in records if r.deleted]
    assert [r.identifier for r in deleted] == ["oai:museum:2"]
    assert deleted[0].metadata is None

    # Selective args ride the FIRST request only; the continuation is verb + token alone.
    assert session.calls[0]["from"] == "2026-07-15"
    assert set(session.calls[1]) == {"verb", "resumptionToken"}
    assert session.calls[1]["resumptionToken"] == "TOKEN-abc"

A green run confirms three things at once: the harvest consumed exactly two pages and halted on the empty token, the status="deleted" record surfaced as a tombstone with no payload, and the continuation request dropped every selective argument down to the verb and token. Those are precisely the three failure modes from the root-cause analysis, now guarded.

Standards Alignment

The harvest conforms to the OAI-PMH 2.0 specification: ListRecords with metadataPrefix, from, until, and set; flow control via an opaque resumptionToken echoed with the verb alone; UTCdatetime bounds; and deleted-record support signalled by a header status="deleted". Every repository must offer oai_dc, mapped from simple Dublin Core, but for museum aggregation the richer LIDO metadataPrefix preserves the structured events, actors, and rights that flat Dublin Core discards — which is why harvested LIDO flows on to parsing LIDO XML with lxml iterparse for element-level extraction. Downstream, the validated records land in the same contract the rest of the ingestion pipeline enforces before any rights or delivery decision is made.

FAQ

How do I know when the harvest is finished?

When the <resumptionToken> element comes back either absent or present but empty. A conformant repository issues an empty token on the final page precisely to signal completion, so the loop’s exit condition is if not token: return. Never treat a fixed record count or a single response as “done” — the first page almost always looks complete while a token is quietly telling you to keep going.

Can I save a resumptionToken and resume the harvest tomorrow?

No. Tokens are stateful, opaque server cursors with a bounded lifetime, and a stale one returns badResumptionToken. Persist the date watermark instead — the until bound of the last clean run — and resume with a fresh from/until window. That is what makes an interrupted harvest safely replayable.

Why did the server reject my continuation request with badArgument?

Because it carried selective arguments alongside the token. Once a resumptionToken is issued, the only legal arguments are the verb and the token — metadataPrefix, from, until, and set must be dropped. Send them only on the very first request of a window.

What happens to records the partner deleted?

They arrive inside the window as records with status="deleted" in the header and no <metadata>. The harvester flags them (deleted=True, metadata=None) and routes them to a purge so the local aggregation copy is removed. If the repository declares deletedRecord support as no, deletions are invisible over OAI-PMH and you must reconcile them with a periodic full re-harvest.