Mert

Handling HTTP 429 Too Many Requests responses correctly in Python data pipelines requires more than a time.sleep(1) in an except block. The following tools and libraries are the practical toolkit for building rate limit resilience into production data automation. They cover everything from simple retry decorators to distributed rate limiting for multi-worker pipelines.

1. Tenacity

What it does: A Python retry library that provides a decorator-based interface for configuring retry behavior, backoff strategies, and logging.

Why it matters: Rolling your own exponential backoff is straightforward for simple cases, but production pipelines need configurable stop conditions, structured logging of retry attempts, and clean separation between business logic and retry behavior. Tenacity handles all of these.

Installation:

pip install tenacity

Basic usage for API rate limits:

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
    before_sleep_log,
)
import logging

logger = logging.getLogger(__name__)

@retry(
    retry=retry_if_exception_type(RateLimitError),
    wait=wait_exponential_jitter(initial=1, max=60),
    stop=stop_after_attempt(6),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def call_api(url, session):
    response = session.get(url, timeout=30)
    if response.status_code == 429:
        raise RateLimitError(response.headers.get("Retry-After"))
    response.raise_for_status()
    return response

Best for: Any production pipeline that makes retried API calls. The before_sleep_log parameter in particular is valuable for operations monitoring — you get a WARNING log entry before every retry, making it easy to set up alerts when retry rates increase.

GitHub: jd/tenacity | Documentation

2. Requests with Session Reuse

What it does: The requests library’s Session object reuses TCP connections across requests, significantly reducing overhead in high-volume API calls.

Why it matters: Each new requests.get() call opens a new TCP connection. At 100 requests per second, this overhead adds up. A Session object maintains a connection pool that reuses established connections, reducing latency and server load.

import requests

with requests.Session() as session:
    session.headers.update({"Authorization": "Bearer TOKEN"})
    session.headers.update({"User-Agent": "MyPipeline/1.0"})

    for url in url_list:
        response = session.get(url, timeout=30)
        # process response

Best for: Any pipeline making multiple calls to the same API host. The combination of session reuse and rate limiting is the baseline for efficient API automation.

Documentation: requests.readthedocs.io

Photo by Brett Sayles on Pexels

3. PyRateLimit / ratelimit

What it does: A simple decorator library for rate limiting function calls. Enforces a maximum number of calls per time period.

Installation:

pip install ratelimit

Usage:

from ratelimit import limits, sleep_and_retry

CALLS_PER_SECOND = 10
ONE_SECOND = 1

@sleep_and_retry
@limits(calls=CALLS_PER_SECOND, period=ONE_SECOND)
def fetch(url, session):
    return session.get(url, timeout=30)

The @sleep_and_retry decorator combined with @limits creates a proactive rate limiter: when the call limit is reached, it sleeps until the window resets rather than raising an exception. This prevents most 429 responses from occurring rather than recovering from them.

Limitations: This library is effective for single-process pipelines but does not coordinate state across multiple workers or processes. For distributed pipelines, you need shared state (see Redis below).

PyPI: pypi.org/project/ratelimit

4. Redis-Based Distributed Rate Limiting

What it does: Uses Redis as a shared rate limit counter across multiple workers, enabling coordinated rate limiting in distributed pipelines.

Why it matters: A token bucket or rate limiter that lives in a single Python process is correct for single-worker pipelines. When you have 10 workers making concurrent requests, each with its own in-process rate limiter, the total request rate will be 10x the per-worker limit. Shared Redis state ensures the total rate is correct regardless of worker count.

Basic pattern using redis-py:

import redis
import time

r = redis.Redis(host="localhost", port=6379)

def acquire_token(key, limit, window_seconds):
    """Sliding window rate limiter backed by Redis."""
    now = time.time()
    pipe = r.pipeline()

    # Remove old entries outside the window
    pipe.zremrangebyscore(key, 0, now - window_seconds)
    # Count current entries
    pipe.zcard(key)
    # Add current request timestamp
    pipe.zadd(key, {str(now): now})
    # Set TTL to clean up automatically
    pipe.expire(key, window_seconds + 1)

    results = pipe.execute()
    current_count = results[1]

    if current_count >= limit:
        return False  # Rate limit exceeded
    return True

Best for: Multi-worker or multi-machine pipelines where coordination of API usage across processes is required.

5. httpx with AsyncClient

What it does: An async HTTP client with request/response API compatible with requests, supporting async/await for concurrent API calls.

Why it matters: asyncio-based pipelines can make many concurrent API calls efficiently without multi-threading overhead. httpx integrates cleanly with tenacity for async retry:

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    wait=wait_exponential_jitter(initial=1, max=60),
    stop=stop_after_attempt(5),
)
async def async_fetch(client, url):
    response = await client.get(url, timeout=30.0)
    if response.status_code == 429:
        raise RateLimitError(response.headers.get("Retry-After"))
    response.raise_for_status()
    return response

async def batch_fetch(urls, headers):
    async with httpx.AsyncClient(headers=headers) as client:
        tasks = [async_fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

Caveat: Async concurrency increases the rate at which you consume API quota. Pair httpx with an async-compatible rate limiter (aiolimiter is one option) to avoid immediately hitting limits with the increased concurrency.

6. Monitoring: Structlog for Retry Visibility

What it does: A structured logging library that produces machine-readable log output, making retry events easier to query and alert on.

Why it matters: Knowing that your pipeline is retrying is only useful if you are notified when retry rates increase. Structlog output integrates cleanly with log aggregation systems (Datadog, CloudWatch, Grafana Loki) where you can set alerts on retry event counts.

import structlog

log = structlog.get_logger()

def log_rate_limit_event(response, wait_seconds):
    log.warning(
        "rate_limit_encountered",
        status_code=response.status_code,
        retry_after=response.headers.get("Retry-After"),
        x_ratelimit_remaining=response.headers.get("X-RateLimit-Remaining"),
        wait_seconds=round(wait_seconds, 1),
    )

Each retry event emits a structured log record with all the context needed to diagnose pattern changes: which endpoint, how long the wait, how many tokens remain. This is the observability layer that turns rate limit handling from a fire-and-forget implementation into a monitored pipeline component.

Photo by Suki Lee on Pexels

Testing Rate Limit Handling

Rate limit handling code is only as good as its test coverage. Three approaches for testing retry logic without hitting a live API:

Mock the response object: Use unittest.mock.patch to replace requests.Session.get with a function that returns a mock response with status_code=429 for the first N calls, then 200 thereafter. This verifies that your retry loop executes the correct number of retries and calls the wait function with the expected arguments.

httpbin: httpbin.org/status/429 returns a real 429 response, letting you verify that your Retry-After parsing and backoff logic work against a live endpoint without consuming your actual API quota.

Local proxy: A local reverse proxy (nginx, mitmproxy) configured to inject 429 responses at a configured rate lets you test the full pipeline behavior — including session reuse, token bucket throttling, and retry logging — under simulated rate limit conditions.

Choosing the Right Combination

For most Python data automation pipelines, the practical starting point is:

• tenacity for retry logic (reactive)
• ratelimit or a hand-rolled token bucket for rate throttling (proactive)
• requests Session for connection efficiency
• Structured logging for observability

For distributed pipelines with multiple workers:

• Add Redis for shared rate limit state across workers
• Consider httpx + asyncio for high-concurrency fetch patterns

The full implementation of these patterns — including a complete working example combining tenacity, token bucket, and session reuse — is covered in How to Handle API Rate Limits in Python Data Automation.

For production data pipeline work where rate limit resilience is a design requirement from day one, visit https://137foundry.com/services/data-automation.

Hitting an API rate limit and not knowing what to do with the HTTP 429 response is one of the most common causes of brittle data automation scripts. This is a step-by-step implementation guide: from a minimal correct backoff function to a production-grade tenacity decorator that logs retries, handles Retry-After headers, and distinguishes between retriable and non-retriable errors.

What We Are Building

By the end of this guide, you will have:

1. A calculate_wait() function that reads Retry-After headers when present and falls back to exponential backoff with jitter when not
2. A fetch_with_backoff() wrapper function for single requests
3. A tenacity-based decorator for production use with logging
4. A proactive TokenBucket class to prevent most 429 responses before they occur

Prerequisites: Python 3.8+, requests and tenacity installed:

pip install requests tenacity

Step 1: Parse the Retry-After Header

The most important piece of rate limit handling is honoring the API’s own signal about when to retry. Many 429 responses include a Retry-After header that tells you exactly how long to wait.

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

def parse_retry_after(headers):
    """
    Parse the Retry-After header from a rate-limited response.
    Returns seconds to wait, or None if the header is absent.
    """
    retry_after = headers.get("Retry-After")
    if not retry_after:
        return None

    # Some APIs return seconds as a plain integer string
    try:
        return max(0.0, float(retry_after))
    except ValueError:
        pass

    # Others return an HTTP date string: "Wed, 21 Oct 2015 07:28:00 GMT"
    try:
        reset_dt = parsedate_to_datetime(retry_after)
        now = datetime.now(timezone.utc)
        return max(0.0, (reset_dt - now).total_seconds())
    except Exception:
        return None

Step 2: Build the Wait Calculation Function

Combine Retry-After parsing with exponential backoff as the fallback:

import random

def calculate_wait(response, attempt, base=1.0, max_delay=60.0):
    """
    Calculate how long to wait before retrying a failed request.
    Uses Retry-After header when available, exponential backoff otherwise.
    """
    api_specified = parse_retry_after(response.headers)
    if api_specified is not None:
        return api_specified

    # Exponential backoff with jitter
    delay = min(base * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.1)
    return delay + jitter

The jitter term prevents the thundering herd problem: if multiple workers all hit the limit at the same moment, jitter ensures they do not all retry at exactly the same moment.

Step 3: Write the Retry Wrapper

A minimal correct implementation with explicit handling for 429 vs. server errors vs. client errors:

import requests

def fetch_with_backoff(url, headers=None, max_retries=6, base_delay=1.0):
    """
    Make a GET request with retry logic for 429 and 5xx responses.
    Does not retry on 4xx client errors (except 429).
    """
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers or {}, timeout=30)

        if response.status_code == 200:
            return response

        if response.status_code == 429:
            if attempt == max_retries - 1:
                raise RuntimeError(f"Rate limit persists after {max_retries} retries: {url}")
            wait = calculate_wait(response, attempt, base=base_delay)
            print(f"Rate limited (429). Waiting {wait:.1f}s (attempt {attempt + 1}/{max_retries})")
            time.sleep(wait)
            continue

        if response.status_code >= 500:
            if attempt == max_retries - 1:
                response.raise_for_status()
            wait = min(base_delay * (2 ** attempt), 60.0)
            print(f"Server error ({response.status_code}). Waiting {wait:.1f}s")
            time.sleep(wait)
            continue

        # 4xx client errors: do not retry
        response.raise_for_status()

    raise RuntimeError(f"Exhausted retries for {url}")

This handles the three distinct cases: rate limits (retry with API-specified or backoff delay), server errors (retry with backoff), and client errors (fail immediately without retrying).

Photo by QuinceCreative on Pixabay

Step 4: Production-Grade Retry with Tenacity

The tenacity library provides a decorator-based retry system that is cleaner to configure and includes built-in logging:

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
    before_sleep_log,
)
import logging
import requests

logger = logging.getLogger(__name__)

class RateLimitError(Exception):
    pass

class ServerError(Exception):
    pass

def raise_for_status_with_retry(response):
    """Convert HTTP errors to typed exceptions for tenacity."""
    if response.status_code == 429:
        raise RateLimitError(
            f"Rate limited. Retry-After: {response.headers.get('Retry-After', 'not specified')}"
        )
    if response.status_code >= 500:
        raise ServerError(f"Server error {response.status_code}")
    response.raise_for_status()
    return response

@retry(
    retry=retry_if_exception_type((RateLimitError, ServerError)),
    wait=wait_exponential_jitter(initial=1, max=60),
    stop=stop_after_attempt(6),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)
def fetch_api_resource(url, session):
    response = session.get(url, timeout=30)
    return raise_for_status_with_retry(response)

The before_sleep_log parameter writes a WARNING entry to your log system before each sleep interval, which makes retry behavior visible in logs without requiring custom logging code in the retry loop.

Step 5: Add Proactive Rate Limiting with Token Bucket

Exponential backoff is reactive: it handles failures after they occur. A token bucket implementation is proactive: it throttles your own request rate to stay below the API limit, reducing how often 429 responses occur.

import threading

class TokenBucket:
    """Thread-safe token bucket for rate limiting API requests."""

    def __init__(self, rate, capacity):
        self.rate = rate          # tokens added per second
        self.capacity = capacity  # maximum tokens
        self.tokens = float(capacity)
        self.last_refill = time.monotonic()
        self._lock = threading.Lock()

    def acquire(self, tokens=1):
        """Wait until tokens are available. Returns the actual wait time."""
        with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return 0.0

            wait = (tokens - self.tokens) / self.rate
            self.tokens = 0
            return wait

# For an API allowing 10 requests per second:
bucket = TokenBucket(rate=10, capacity=10)

def throttled_fetch(url, session):
    wait = bucket.acquire()
    if wait > 0:
        time.sleep(wait)
    return fetch_api_resource(url, session)

Putting It Together

A complete paginated API consumer that combines proactive throttling and reactive retry:

import requests

bucket = TokenBucket(rate=5, capacity=10)  # Stay under the API limit

def paginate_api(base_url, auth_headers, params=None):
    results = []
    page = 1

    with requests.Session() as session:
        session.headers.update(auth_headers)

        while True:
            url = f"{base_url}?page={page}&per_page=100"
            response = throttled_fetch(url, session)

            data = response.json()
            items = data.get("items") or data.get("results") or []

            if not items:
                break

            results.extend(items)
            page += 1

    return results

The token bucket prevents most 429 responses. The tenacity decorator handles the ones that slip through. The session reuses the TCP connection across requests.

Error Categories to Handle Differently

Not all errors should be retried. A well-structured retry strategy distinguishes:

Retrying on 401 or 400 is always wrong and wastes time. A correct implementation routes each error type to the appropriate response.

Photo by blickpixel on Pixabay

Testing Your Retry Logic

Retry logic is difficult to test without a server that deliberately returns 429 responses. Options:

Mock the response object: Use unittest.mock.patch to replace requests.get with a function that returns a mock response with status_code=429 for the first N calls.

Use a local proxy: Run a local reverse proxy (nginx, mitmproxy) in front of the API that injects 429 responses at a configured rate.

Use httpbin: httpbin.org/status/429 returns a 429 response that you can use to test parsing and backoff behavior without hitting a real API.

For the complete guide including token bucket implementation details, tenacity configuration for production pipelines, and monitoring recommendations, read the full article here.