Consuming REST APIs Cleanly

When the API Is the Front Door#

Not every data source needs HTML parsing or reverse engineering. Many services publish a documented REST API, and many web apps call JSON endpoints that are clean enough to use directly. An API gives you structured data, stable field names and explicit rules about rate limits. Consuming one well is still a skill: authenticate correctly, page through every record without gaps or duplicates, stay inside the rate limit, recover from transient errors, and fetch only what changed. This lesson builds a small, reusable API client that does all of that.

Read the Contract First#

Before writing code, spend ten minutes on the documentation and look for five things: how to authenticate, the rate limits, the pagination style, the maximum page size, and any filter for "changed since". Also check the terms of use; API terms often restrict caching, redistribution or commercial use. Many APIs publish an OpenAPI (Swagger) description, often at a path like /openapi.json, which lists every endpoint, parameter and response schema in machine-readable form and is faster to search than prose documentation.

For undocumented endpoints behind a web app, found with the techniques in XHR and API reverse engineering, everything below still applies, with one addition: they can change without notice, so validate responses strictly.

Build Requests Correctly#

REST endpoints use two kinds of parameters. Path parameters identify a resource (/users/42); query parameters filter or shape the response (?fields=name,email&page=2). Pass query parameters as a dict so the client encodes them, and encode path segments yourself:

import os
from urllib.parse import quote
import httpx

client = httpx.Client(
    base_url="https://api.example.com/v2",
    headers={
        "Authorization": f"Bearer {os.environ['EXAMPLE_API_TOKEN']}",
        "Accept": "application/json",
        "User-Agent": "acme-catalog-sync/1.0 (+data-team@acme.example)",
    },
    timeout=httpx.Timeout(20.0, connect=5.0),
)

sku = "LAMP/42 blue"
r = client.get(f"/products/{quote(sku, safe='')}", params={"fields": "id,title,price", "locale": "en"})
print(r.request.url)

https://api.example.com/v2/products/LAMP%2F42%20blue?fields=id%2Ctitle%2Cprice&locale=en

quote(..., safe="") matters for IDs that contain /: without it, LAMP/42 becomes two path segments and hits a different endpoint. Never build query strings with f-strings; a search term containing & or # silently changes the request. Keep tokens in environment variables or a secrets manager, never in code, and prefer header-based keys over ?api_key= query parameters, which end up in server logs and proxy logs.

Authentication Patterns#

For client credentials, cache the token and renew it shortly before it expires:

import time

class TokenProvider:
    def __init__(self, token_url, client_id, client_secret):
        self.token_url, self.auth = token_url, (client_id, client_secret)
        self._token, self._expires_at = None, 0.0

    def get(self) -> str:
        if self._token is None or time.time() > self._expires_at - 60:
            r = httpx.post(self.token_url, data={"grant_type": "client_credentials"},
                           auth=self.auth, timeout=15)
            r.raise_for_status()
            body = r.json()
            self._token = body["access_token"]
            self._expires_at = time.time() + float(body.get("expires_in", 3600))
        return self._token

The 60-second margin avoids sending a token that expires while the request is in flight.

Rate Limits, Read From the Response#

Well-behaved APIs tell you your budget in response headers. The names vary: X-RateLimit-Limit, X-RateLimit-Remaining and X-RateLimit-Reset are common, and some APIs use the RateLimit and RateLimit-Policy headers from an IETF draft. X-RateLimit-Reset is a Unix timestamp in some APIs and a number of seconds in others. Retry-After, sent with 429 and 503 responses, is either a number of seconds or an HTTP date. Handle all of these:

import random
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

def retry_after_seconds(r):
    value = r.headers.get("retry-after")
    if not value:
        return None
    if value.strip().isdigit():
        return float(value)
    try:
        when = parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())

def pace(r):
    remaining = r.headers.get("x-ratelimit-remaining")
    reset = r.headers.get("x-ratelimit-reset")
    if remaining is None or reset is None:
        return
    remaining, reset = float(remaining), float(reset)
    seconds_left = reset - time.time() if reset > 1e9 else reset    # timestamp or delta
    if remaining <= 0:
        time.sleep(max(0.0, seconds_left) + 1)
    elif remaining < 10 and seconds_left > 0:
        time.sleep(seconds_left / remaining)        # spread the last requests over the window

Proactive pacing is better than reacting to 429s. When the remaining budget is low, spacing requests evenly until the reset keeps you under the limit instead of burning the budget and then stalling. If an API publishes no headers, use the documented limit and a token bucket from rate limiting and politeness, staying comfortably below it.

Retry the Right Failures#

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

def get_json(client, path, params=None, attempts=5):
    for attempt in range(attempts):
        try:
            r = client.get(path, params=params)
        except httpx.TransportError as exc:          # timeouts, connection resets, DNS
            wait, error = min(60, 2 ** attempt) + random.random(), exc
        else:
            if r.status_code < 400:
                pace(r)
                return r
            if r.status_code not in RETRYABLE:
                detail = r.text[:300]
                raise RuntimeError(f"{r.status_code} on {r.request.url}: {detail}")
            wait = retry_after_seconds(r) or min(60, 2 ** attempt) + random.random()
            error = RuntimeError(f"{r.status_code} after {attempts} attempts")
        time.sleep(wait)
    raise error

• 4xx errors other than 429 are not retried. A 400 or 422 means your request is wrong; a 403 means you are not allowed; a 404 means the resource does not exist. Repeating them wastes budget. A 401 is the one exception worth handling specially: refresh the token once and try again.
• Include the start of the error body in the exception. APIs usually explain what is wrong ("per_page must be <= 100"), and that message saves a debugging session.
• Log any request ID header the API returns (often X-Request-ID); support teams can trace a specific failing call with it.

The general taxonomy of retryable errors is in error handling.

Pagination Without Gaps or Duplicates#

APIs paginate in four common styles: offset and limit (?offset=100&limit=50), page numbers (?page=3&per_page=50), cursors (the response contains a next_cursor token), and Link headers (Link: <...?page=4>; rel="next"). One generator can handle cursors and link headers, with guards against infinite loops:

def paginate(client, path, params=None, max_pages=10_000):
    params = dict(params or {})
    url, seen = path, set()
    for _ in range(max_pages):
        r = get_json(client, url, params)
        data = r.json()
        yield from data["items"]

        next_url = r.links.get("next", {}).get("url")
        cursor = data.get("next_cursor")
        if next_url:
            url, params = next_url, None              # next link already contains the query
        elif cursor:
            params = {**(params or {}), "cursor": cursor}
        else:
            return
        token = next_url or cursor
        if token in seen:
            raise RuntimeError(f"pagination repeated {token!r}; stopping")
        seen.add(token)
    raise RuntimeError("max_pages reached; check for a pagination bug")

r.links parses the Link header into a dict keyed by rel. Setting params to None for a next link is deliberate. Since httpx 0.28, passing a params dict, even an empty one, replaces the query string already in the URL instead of merging with it, which would drop the page number the server put there (httpx 0.27 and earlier merged the two, so the same code behaves differently across versions). The same replacement happens in 0.28 if the Client itself was created with default params=, even when the request passes params=None, so keep default query parameters off a client that follows next links. The seen set and max_pages protect against APIs that return the same cursor forever, which would otherwise hammer the host until someone notices.

Offset pagination has two traps, covered in more depth in pagination:

• Shifting data. If records are inserted while you page, items slide between pages, causing duplicates and misses. Sort by a stable key (sort=id) and deduplicate by ID.
• Depth caps. Many APIs refuse offsets beyond a limit, so "page 500" returns an error or empty results even though total says there are more. Split the query into slices with filters, such as date ranges, so each slice stays under the cap, and split a slice again if it hits the cap.

from datetime import date, timedelta

def date_slices(start: date, end: date, days: int = 7):
    cur = start
    while cur < end:
        nxt = min(cur + timedelta(days=days), end)
        yield cur, nxt
        cur = nxt

for lo, hi in date_slices(date(2025, 1, 1), date(2026, 1, 1)):
    for order in paginate(client, "/orders", {"created_from": lo.isoformat(),
                                              "created_to": hi.isoformat(), "sort": "id"}):
        store(order)

Fetch Less: Fields, Page Size and Change Filters#

The cheapest request is the one you do not make:

• Use the maximum page size the API allows. Fifty pages of 100 cost half the requests and budget of 100 pages of 50.
• Request only the fields you need if the API supports fields= or sparse fieldsets; responses get smaller and faster.
• Use change filters for repeat syncs. With an updated_since parameter, store the highest updated_at you have seen (a watermark) and request from slightly before it next time. The overlap of a few minutes covers clock skew and records committed late; deduplicating by ID makes the overlap harmless.
• Use conditional requests where supported: send If-None-Match with the stored ETag and treat 304 Not Modified as "nothing to do". The mechanics are in incremental scraping.

Validate What Comes Back#

A 200 response is not proof of good data. Check that the fields you depend on exist and have the right types, count records per run against the API's total, and store the raw JSON for a sample of responses so you can re-parse later if you find a bug. The extraction side of nested responses is covered in nested JSON payloads.

Practice#

Pick a free public API with documented rate limits and cursor or link pagination. Implement get_json and paginate, run a full sync, and log the rate-limit headers from every response. Then run it again with an updated_since watermark and compare the request counts of the full and incremental runs.