Data Ingestion Quick Reference

A compact lookup for requests sessions, retry and backoff, cursor pagination, auth headers, webhook signature verification, poll-until-ready jobs, and mocked HTTP tests.

BeginnerMay 18, 2026

Before you use this reference

This page is a lookup, not a tutorial. The explanations live in Calling REST APIs with Python, Pagination and Rate Limiting, API Authentication Patterns, Webhooks and Event-Driven Ingestion, Bulk Export APIs and the Poll-Until-Ready Pattern, and Testing Ingestion Code Without Hitting the Network.

Session, timeout, error handling#

Full reference: requests.

Python

import requests
session = requests.Session()session.headers.update({"Authorization": f"Bearer {api_key}"})
response = session.get(url, params=params, timeout=(3.05, 10))response.raise_for_status()data = response.json()

Piece	Why
`Session()`	Reuses the TCP connection pool across calls
`timeout=(connect, read)`	Never left unset — an unset timeout can hang forever
`raise_for_status()`	Turns a 4xx/5xx response into an exception before `.json()` runs

Status code decision#

Text

2xx            success4xx (not 429)  client error — do not retry unchanged429            rate limited — respect Retry-After, or back off5xx            server error — retry with backoff

Full taxonomy: List of HTTP status codes.

Retry with exponential backoff and jitter#

Python

import randomimport time
def call_with_backoff(fn, max_attempts=5, base_delay=1.0):    for attempt in range(max_attempts):        try:            return fn()        except requests.HTTPError as exc:            status = exc.response.status_code            if status < 500 and status != 429:                raise            if attempt == max_attempts - 1:                raise            time.sleep(base_delay * (2 ** attempt) + random.uniform(0, 1))

Cursor pagination loop#

Python

def fetch_all(client, path, **params):    results = []    cursor = None    while True:        page = client.get(path, params={**params, "cursor": cursor}).json()        results.extend(page["results"])        cursor = page.get("next_cursor")        if cursor is None:            break    return results

Prefer a cursor over an offset/page-number scheme whenever the source offers one — offset pagination skips or duplicates rows once the underlying table is written to while you are still paging through it.

Auth headers#

Scheme	Header	Expires
API key	`X-API-Key: <key>` (name varies by source)	Rarely
Basic Auth	`Authorization: Basic <base64(user:pass)>`	No
OAuth 2.0 bearer	`Authorization: Bearer <token>`	Yes — cache and refresh near expiry

Python

# OAuth 2.0 client credentials grantresponse = requests.post(token_url, data={    "grant_type": "client_credentials",    "client_id": client_id,    "client_secret": client_secret,}, timeout=(3.05, 10))token = response.json()["access_token"]

Webhook signature verification#

Python

import hashlibimport hmac
def verify_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:    expected = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()    return hmac.compare_digest(expected, signature_header)

Always sign and compare the raw request body, not a re-serialized copy of the parsed JSON — re-serialization can reorder keys and break the signature for a genuine payload.

Idempotent event handling skeleton#

Python

def process_event(payload: dict) -> None:    event_id = payload["event_id"]    if already_processed(event_id):        return    apply_event(payload)    mark_processed(event_id)

Poll-until-ready skeleton#

Python

def wait_for_export(client, job_id, poll_interval=10, timeout=1800):    deadline = time.time() + timeout    while time.time() < deadline:        status = client.get(f"/exports/{job_id}").json()        if status["state"] in ("completed", "failed"):            return status        time.sleep(poll_interval)    raise TimeoutError(f"Export {job_id} did not finish within {timeout}s")

Check state on every poll — a "failed" job is a terminal state, not "still running."

Mocked HTTP test skeleton#

Python

import responses
@responses.activatedef test_get_orders():    responses.add(        responses.GET, "https://api.example.com/orders",        json={"results": [], "next_cursor": None}, status=200,    )    assert client.get_orders() == []

Data Ingestion Quick Reference

Session, timeout, error handling#

Status code decision#

Retry with exponential backoff and jitter#

Auth headers#

Webhook signature verification#

Idempotent event handling skeleton#

Poll-until-ready skeleton#

Mocked HTTP test skeleton#

Further reading#

See also#