Operational Context

A Python automation engineer stands up a synchronous polling script to mirror a museum CMS into a downstream data warehouse, expecting near-real-time copies of object metadata, rights statements, and IIIF manifest URLs — and instead watches the job collapse under load. Cascading HTTP 429 and 503 responses stall worker threads, hardcoded time.sleep() intervals either throttle throughput or trip immediate backoff, and in-memory pagination tracking drifts on the first restart, producing duplicate records, orphaned media references, and broken provenance chains. This page resolves that exact failure: rebuilding an ad-hoc requests.get() loop into a deterministic engine that respects API rate limits, holds a constant memory footprint, and survives process restarts without re-ingesting the collection. It is the synchronous entry point into the asynchronous ingestion pipeline, and it feeds the same validated-record contract the rest of the automated record ingestion pipeline depends on.

Root Cause Analysis

Museum API instability originates from three intersecting failure vectors. First, repeated requests.get() calls open and close a fresh TCP socket each time, exhausting ephemeral ports and starving connection pools under concurrency. Second, sliding-window rate limits advertised through X-RateLimit-Remaining and Retry-After headers are ignored in favour of fixed sleeps, so the client either wastes its quota or hammers straight into a backoff wall. Third, loading a complete JSON array into a Python list scales memory linearly with collection size, and constrained CI runners trigger OOM kills partway through a bulk metadata harvest.

Schema drift compounds all three. Missing ObjectID fields, malformed ISO 8601 timestamps, and unescaped MARC fragments break downstream ETL the moment they land, so validation has to happen at the ingestion boundary rather than after a database commit — the same discipline enforced in schema validation with Pydantic. Left unaddressed, these vectors turn a nightly sync into a state-drifted loop that requires manual reconciliation to recover.

Canonical Solution

Resolving the instability means restructuring the request lifecycle around four invariants: a persistent session that reuses TCP connections, transport-level retry that honours rate-limit headers, streaming payload processing that keeps memory flat, and durable filesystem checkpoints that make the cursor survive a crash. The diagram below shows how those pieces compose into a single deterministic cycle.

The deterministic museum-API poll cycle A closed loop. Load checkpoint feeds GET page (session plus retry), which streams JSON lines into a Validate record decision. The ok branch flows right to Sync record (advance cursor); the fail branch drops to Log and skip. Both converge on Save checkpoint, then a More pages decision loops the yes branch back around the left edge to GET page and sends the no branch down to Cycle complete. ok fail yes no Load checkpoint cursor + last_sync GET page session + retry Stream JSON lines iter_lines() Validate record Sync record advance cursor Log + skip row never halts run Save checkpoint cursor to disk More pages? Cycle complete

Step 1: Session Pooling and Adaptive Backoff

Initialize a persistent requests.Session to reuse TCP connections across every request, then attach urllib3 retry logic at the transport-adapter level so transient drops recover without any application-level branching. Setting respect_retry_after_header=True makes the client obey the exact wait the API dictates — whether that arrives as a delay in seconds or as an HTTP-date — instead of guessing with a fixed sleep. The status_forcelist targets precisely the codes a loaded museum API returns under pressure.

python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from typing import Iterator
import logging

logger = logging.getLogger(__name__)

def configure_session(max_retries: int = 3) -> requests.Session:
    session = requests.Session()
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1.0,            # 1s, 2s, 4s between retries
        status_forcelist=[429, 502, 503, 504],
        allowed_methods=["GET"],
        respect_retry_after_header=True
    )
    adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=20)
    session.mount("https://", adapter)
    session.headers.update({"Accept": "application/json", "User-Agent": "MuseumSync/1.0"})
    return session

Step 2: Streaming Payloads and Memory Boundaries

Never buffer the full response body. Iterate with response.iter_lines() so records are processed one at a time and peak memory stays constant regardless of collection size. The reader below assumes the endpoint streams newline-delimited JSON (JSON Lines) — the common shape for bulk harvesting endpoints; for an endpoint that returns a single large JSON array, swap in an incremental parser such as ijson. Consuming the response inside a with block guarantees the connection returns to the pool even if iteration is abandoned mid-stream, which is what keeps the pool from leaking during partial reads. This mirrors the streaming strategy used to keep bulk CSV syncs flat in handling large CSV batches without memory leaks.

python
import json

def stream_records(session: requests.Session, endpoint: str, params: dict[str, str]) -> Iterator[dict]:
    with session.get(endpoint, params=params, stream=True) as response:
        response.raise_for_status()
        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as e:
                # One malformed line must not abort the whole page.
                logger.warning("Malformed JSON payload skipped: %s", e)

Step 3: Schema Validation at the Ingestion Boundary

Enforce structural compliance before a record reaches the persistence layer. A Pydantic v2 model validates required fields, coerces types, and rejects malformed timestamps up front, so downstream ETL never sees a record it cannot map. The ObjectID alias bridges the CMS field name to a Python-friendly attribute, and the date validator normalizes the Z suffix so datetime.fromisoformat parses cleanly. Non-conforming records fail here rather than at a database constraint, and mapping the accepted record onto a LIDO structure is handed off to the LIDO-to-database mapping layer.

python
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional

class LIDORecord(BaseModel):
    object_id: str = Field(alias="ObjectID")
    title: str
    creation_date_iso: Optional[str] = None
    iiif_manifest_url: Optional[str] = None

    @field_validator("creation_date_iso")
    @classmethod
    def validate_iso8601(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        # Reject anything fromisoformat cannot parse; "Z" -> "+00:00" first.
        datetime.fromisoformat(v.replace("Z", "+00:00"))
        return v

def validate_record(raw: dict) -> LIDORecord:
    return LIDORecord.model_validate(raw)

Step 4: Checkpointing and State Drift Prevention

Track the pagination cursor and last-synced timestamp in durable storage, not in a variable that dies with the process. Writing a checkpoint after every successful batch means a crash resumes from the last verified cursor instead of restarting the harvest — the single change that eliminates duplicate ingestion and the provenance-chain breaks it causes. A local JSON manifest is enough for a single-node poller; a shared run would promote this to a row in the CMS or a key in Redis.

python
from pathlib import Path

CHECKPOINT_FILE = Path("sync_state.json")

def load_checkpoint() -> dict[str, str]:
    if CHECKPOINT_FILE.exists():
        return json.loads(CHECKPOINT_FILE.read_text())
    return {"cursor": "0", "last_sync": datetime.now().isoformat()}

def save_checkpoint(cursor: str) -> None:
    state = {"cursor": cursor, "last_sync": datetime.now().isoformat()}
    CHECKPOINT_FILE.write_text(json.dumps(state, indent=2))

Step 5: The Deterministic Poll Loop

The orchestration loop composes the four steps into one cycle. Each record crosses the validation boundary before it counts as synced; a failed validation is logged and skipped so a single malformed row never halts the run, and wiring that except branch to a durable quarantine queue is a small extension of the pattern documented for Celery-based museum data sync. The loop terminates cleanly when a page returns nothing or the cursor stops advancing, so it can never spin on an exhausted endpoint.

python
def run_sync(session: requests.Session, endpoint: str) -> None:
    checkpoint = load_checkpoint()
    cursor = checkpoint.get("cursor", "0")

    while True:
        params = {"cursor": cursor, "limit": "500"}
        page_cursor = cursor
        received = False
        for record in stream_records(session, endpoint, params):
            received = True
            try:
                validated = validate_record(record)
                logger.info("Synced record: %s", validated.object_id)
                cursor = record.get("next_cursor", cursor)
            except Exception:
                logger.error("Validation failed for record: %s", record.get("ObjectID"))
                continue
        save_checkpoint(cursor)
        # Stop when a page returns nothing or the cursor stops advancing.
        if not received or cursor == page_cursor:
            break

    logger.info("Sync cycle complete. Next cursor: %s", cursor)

Edge Cases and Variants

  • JSON Lines vs. single array. stream_records assumes newline-delimited JSON. For an endpoint that returns one large {"results": [...]} array, replace iter_lines() with an incremental ijson.items(response.raw, "results.item") parse so memory still stays flat.
  • Cursor vs. offset vs. timestamp pagination. Opaque cursors are safest against concurrent writes. Offset pagination (?page=N) silently skips or repeats records if the collection changes mid-harvest; timestamp windows (?modified_since=) suit incremental syncs but require a small overlap to tolerate clock skew.
  • Retry-After as seconds vs. HTTP-date. respect_retry_after_header=True handles both forms, but if you parse the header manually for custom logic, account for the RFC 7231 HTTP-date variant, not just an integer.
  • Strict vs. lenient validation. In a backfill, catch pydantic.ValidationError, attach the raw payload, and route it to quarantine; in an interactive re-sync, surface the error to the operator instead of silently skipping the row.
  • Auth token refresh. For OAuth2 endpoints, refresh the bearer token before expiry inside configure_session (or a per-request hook) rather than letting a mid-harvest 401 masquerade as a transient failure the retry layer will pointlessly re-attempt.
  • Migrating to async. Once a single poller saturates its rate budget, the same session/validation/checkpoint contract ports directly onto the semaphore-bounded httpx design in building async ingestion pipelines.

Validation

Confirm the two invariants that keep the sync clean — validation rejects malformed input, and the cursor survives a restart — with an assert-based test that needs no live API:

python
# test_polling_sync.py  ->  run with:  pytest -q test_polling_sync.py
import pytest
from pydantic import ValidationError

def test_missing_object_id_is_rejected():
    with pytest.raises(ValidationError):
        validate_record({"title": "Object with no accession number"})

def test_iso_date_is_accepted_and_zulu_normalized():
    rec = validate_record({"ObjectID": "1979.5.1", "title": "Boating",
                           "creation_date_iso": "1874-01-01Z"})
    assert rec.object_id == "1979.5.1"          # aliased from ObjectID

def test_malformed_date_is_rejected():
    with pytest.raises(ValidationError):
        validate_record({"ObjectID": "X", "title": "Y", "creation_date_iso": "June 1889"})

def test_checkpoint_survives_a_restart(tmp_path, monkeypatch):
    import __main__ as m                         # module under test (illustrative)
    monkeypatch.setattr(m, "CHECKPOINT_FILE", tmp_path / "sync_state.json")
    m.save_checkpoint("500")
    assert m.load_checkpoint()["cursor"] == "500"   # cursor persisted to disk

A green run proves that a record without an ObjectID or with an unparseable date never enters the persistence path, and that a resumed process reads back the exact cursor the previous run committed rather than restarting the harvest from zero.

Standards Alignment

The validated payload is deliberately shaped for downstream conformance rather than treated as opaque JSON. The LIDORecord fields map onto a LIDO object record — object_id to the persistent identifier and creation_date_iso to a normalized lido:eventDate — so the harvested record slots into a LIDO pipeline without post-processing, and reconciling free-text places or agents against Getty AAT and TGN upgrades them to stable authority URIs. Where the record carries an iiif_manifest_url, it must resolve to a manifest conforming to the IIIF Presentation API 3.0, whose id is a dereferenceable HTTP(S) URI; field-level structure follows the LIDO schema. Enforcing these at the ingestion boundary is what keeps a fast synchronous poller from quietly filling the CMS with records that fail validation only once a curator opens them.

Frequently Asked Questions

Why does a persistent Session matter more than tuning the sleep interval?

Because the dominant cost under load is TCP and TLS setup, not the wait between requests. Reusing one requests.Session keeps connections warm in the pool, and delegating retry timing to urllib3 with respect_retry_after_header=True means the API — not a hardcoded time.sleep() — decides the backoff. That combination removes both the port-exhaustion failures and the rate-limit whiplash a fixed interval causes.

How do I stop a restart from re-ingesting the whole collection?

Persist the pagination cursor to durable storage after every successful batch, as in save_checkpoint, and read it back with load_checkpoint at startup. In-memory tracking is the direct cause of duplicate records and broken provenance chains after a crash; a cursor on disk (or in the CMS/Redis for shared runs) makes the harvest resumable.

When should I move from Requests to an async client?

When a single synchronous poller cannot keep the collection current within its rate budget, or when you need many endpoints polled concurrently. The session, validation, and checkpoint contract carries over unchanged onto the semaphore-bounded httpx design in the parent async pipeline, so the migration is a transport swap rather than a rewrite.