API Automation: Retries & Rate Limits

By the end of this lesson you can:

• Tell transient errors (retry) from permanent ones (don't).
• Implement exponential backoff with jitter.
• Honour HTTP 429 and the Retry-After header.
• Build a reusable resilient requests.Session for all your API calls.

Over a long run, things will go wrong:

ConnectionError   wifi blipped for a second
500/502/503       server briefly overloaded
429               you're calling too fast — slow down
timeout           the response is taking too long
404 / 401         permanent: wrong URL / bad auth — retrying won't help

A basic GET with JSON

import requests

resp = requests.get("https://api.example.com/v1/users", timeout=10)
resp.raise_for_status()
data = resp.json()      # parse the JSON body into Python

Exponential backoff with jitter

import time, random, requests, logging

log = logging.getLogger("api")
RETRYABLE = {429, 500, 502, 503, 504}

def get_with_retry(url, *, max_attempts=5, timeout=10, **kwargs):
    for attempt in range(1, max_attempts + 1):
        try:
            resp = requests.get(url, timeout=timeout, **kwargs)
            if resp.status_code in RETRYABLE:
                raise requests.HTTPError(f"status {resp.status_code}",
                                         response=resp)
            resp.raise_for_status()        # raise on other 4xx (permanent)
            return resp
        except (requests.ConnectionError, requests.Timeout,
                requests.HTTPError) as e:
            permanent = (isinstance(e, requests.HTTPError)
                         and e.response is not None
                         and e.response.status_code not in RETRYABLE)
            if permanent or attempt == max_attempts:
                raise                       # give up
            wait = min(2 ** (attempt - 1), 30) + random.uniform(0, 1)
            log.warning("attempt %d failed (%s); retrying in %.1fs",
                        attempt, e, wait)
            time.sleep(wait)

• Exponential: wait 1s, 2s, 4s, 8s… (2 ** (attempt-1)), capped at 30s.
• Jitter: + random.uniform(0, 1) spreads retries so clients don't all hit at once.
• Permanent vs transient: a 404/401 raises immediately; a 503/timeout retries.

Honour 429 and Retry-After

def respect_rate_limit(resp) -> float | None:
    if resp.status_code == 429:
        retry_after = resp.headers.get("Retry-After")
        if retry_after:
            return float(retry_after)      # server told us exactly how long
        return 60.0                        # sensible default
    return None

A 429 means "too many requests." The server often includes a Retry-After header with the exact seconds to wait — always prefer that over guessing. Honouring it keeps your API access from being throttled or banned.

The library shortcut

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def resilient_session() -> requests.Session:
    retry = Retry(total=5, backoff_factor=1,
                  status_forcelist=[429, 500, 502, 503, 504],
                  respect_retry_after_header=True)
    s = requests.Session()
    s.mount("https://", HTTPAdapter(max_retries=retry))
    return s

session = resilient_session()
data = session.get("https://api.example.com/v1/data", timeout=10).json()

This gives you backoff + 429 handling in a few lines, applied to every request through the session — the production default once you understand what it's doing.

Goal: pull every page of a paginated API into a CSV, surviving transient errors and respecting rate limits — the canonical "sync the data" automation.

import csv, time, logging
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("sync")

def session() -> requests.Session:
    retry = Retry(total=5, backoff_factor=1,
                  status_forcelist=[429, 500, 502, 503, 504],
                  respect_retry_after_header=True)
    s = requests.Session()
    s.mount("https://", HTTPAdapter(max_retries=retry))
    s.headers.update({"User-Agent": "DataSync/1.0"})
    return s

def fetch_all(base_url: str) -> list[dict]:
    s = session()
    results, page = [], 1
    while True:
        resp = s.get(base_url, params={"page": page, "per_page": 100},
                     timeout=10)
        resp.raise_for_status()
        body = resp.json()
        items = body.get("data", [])
        if not items:
            break                          # no more pages
        results.extend(items)
        log.info("page %d: +%d (total %d)", page, len(items), len(results))
        if not body.get("has_more"):       # API tells us when done
            break
        page += 1
        time.sleep(0.5)                    # gentle pacing on top of retries
    return results

def run(base_url: str, out: str) -> None:
    records = fetch_all(base_url)
    if not records:
        log.warning("no records returned")
        return
    fields = list(records[0].keys())
    with open(out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader(); w.writerows(records)
    log.info("synced %d records → %s", len(records), out)

run("https://api.example.com/v1/orders", "orders.csv")

INFO page 1: +100 (total 100)
WARNING Retrying (Retry(total=4, ...)) after 1s: /v1/orders   ← transient blip, auto-recovered
INFO page 2: +100 (total 200)
INFO page 3: +37 (total 237)
INFO synced 237 records → orders.csv

Read the code

The resilient Session handles retries and 429s transparently — notice the warning line where a blip was auto-recovered mid-run without losing data. On top of that we add gentle sleep(0.5) pacing and use the API's own has_more flag to stop cleanly. The output is a CSV, so everything from Lesson 15 onward applies. This pattern — paginate, retry, rate-limit, write — is how you reliably sync data from any API, and it's a building block for the daily report bot in Lesson 40.

Use a free, no-auth practice API such as https://jsonplaceholder.typicode.com or https://httpbin.org (which can even simulate errors, e.g. /status/503).

# httpbin always returns 503 here, so retries exhaust and raise.
try:
    get_with_retry("https://httpbin.org/status/503", max_attempts=4)
except requests.HTTPError as e:
    print("gave up:", e)

# 404 should raise on the FIRST attempt (no retries),
# 503 should retry up to max_attempts.
for code in (404, 503):
    try:
        get_with_retry(f"https://httpbin.org/status/{code}",
                       max_attempts=3)
    except requests.HTTPError as e:
        print(code, "→", e)

Real API automation must survive transient failures. Retry transient errors (connection drops, timeouts, 5xx, 429) but give up immediately on permanent ones (404, 401). Use exponential backoff (1s, 2s, 4s…) with jitter and a cap, and honour 429's Retry-After header. A requests.Session with the urllib3 Retry adapter gives you all of this in a few lines for every call. Add gentle pacing, paginate to the end, and write results to CSV/JSON. These habits turn a fragile demo into a pipeline you can leave running unattended.