Consuming Real APIs with Resilience: Rate Limits, Pagination, and Fallbacks

A pipeline that calls an external API is a pipeline with a dependency it does not control. The API may be rate-limited, paginated, temporarily unavailable, or inconsistent in what it returns across calls. Writing the happy path is straightforward. Writing the code that handles every failure mode that will actually occur in production is what takes the most time and produces the most value.

The patterns I describe here come from real integrations: CoinGecko for cryptocurrency market data, Open-Meteo for weather, and OpenSky for flight position data. Each of these is a public API with different constraints, and each required a different combination of the same underlying techniques.

Rate limits: respecting the contract

A rate limit is the API provider's way of preventing any single consumer from degrading the service for others. Hitting a rate limit and crashing the pipeline is an engineering failure: the limit is documented, the HTTP 429 response is standardized, and the correct behavior (wait and retry) is well-established. The only question is how long to wait.

import time
import requests

def get_coingecko_market_data(coin_ids: list[str], vs_currency: str = "usd") -> list[dict]:
    """
    CoinGecko free tier: 10-30 requests per minute depending on endpoint.
    Strategy: check Retry-After header if present, otherwise use exponential backoff.
    """
    url = "https://api.coingecko.com/api/v3/coins/markets"
    params = {
        "vs_currency": vs_currency,
        "ids": ",".join(coin_ids),
        "order": "market_cap_desc",
        "per_page": 250,
    }

    for attempt in range(1, 5):
        response = requests.get(url, params=params, timeout=30)

        if response.status_code == 200:
            return response.json()

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
            print(f"Rate limited. Waiting {retry_after}s (attempt {attempt}/4)")
            time.sleep(retry_after)
            continue

        response.raise_for_status()

    raise RuntimeError("Exhausted retries after rate limiting")

The Retry-After header, when present, is more reliable than any backoff formula: the API is telling you exactly how long to wait. Using it respects the provider's intended usage pattern rather than guessing a wait time that may be too short (another 429 immediately) or too long (unnecessary pipeline delay).

For APIs without a Retry-After header, adding jitter to exponential backoff prevents synchronized retries from multiple pipeline instances hitting the API at the same moment:

import random

def smart_backoff(attempt: int) -> float:
    base = min(60, 2 ** attempt)
    return base + random.uniform(0, base * 0.3)

# Attempt 1: 2.0-2.6s, Attempt 2: 4.0-5.2s, Attempt 3: 8.0-10.4s

Pagination: collecting all the data

Most APIs paginate large result sets. The two common approaches are offset-based pagination (page number and page size) and cursor-based pagination (a token returned with each response that identifies where the next page starts). Cursor-based is more reliable for large or frequently-updated datasets: offset-based pagination can miss records or return duplicates if the underlying data changes between page requests.

def fetch_all_flights_opensky(bbox: tuple) -> list[dict]:
    """
    OpenSky Network returns all current flights in a bounding box.
    The API is not paginated for live data, but historical queries are.
    This pattern handles time-windowed pagination for historical pulls.
    """
    lat_min, lon_min, lat_max, lon_max = bbox
    url = "https://opensky-network.org/api/flights/all"

    all_flights = []
    # OpenSky limits to 2-hour windows for historical queries
    window_hours = 2
    current_start = start_time  # defined by caller

    while current_start < end_time:
        current_end = min(current_start + window_hours * 3600, end_time)

        response = requests.get(url, params={
            "begin": current_start,
            "end": current_end,
        }, timeout=60)

        if response.status_code == 200:
            batch = response.json() or []
            all_flights.extend(batch)
            print(f"Fetched {len(batch)} flights for window ending {current_end}")

        current_start = current_end
        time.sleep(1)  # respect the free tier rate limit

    return all_flights

For cursor-based pagination, the loop structure is slightly different: continue until the API returns a null or missing cursor, or an empty page:

def fetch_paginated_cursor(base_url: str, params: dict) -> list[dict]:
    results = []
    cursor = None

    while True:
        if cursor:
            params["cursor"] = cursor

        response = requests.get(base_url, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        page_results = data.get("data", [])
        results.extend(page_results)

        cursor = data.get("next_cursor")
        if not cursor or not page_results:
            break

        time.sleep(0.5)  # courteous pause between pages

    return results

Fallbacks: running when the source does not

Open-Meteo is one of the most reliable free weather APIs available, but no API has 100% availability. A pipeline that has no fallback strategy for source unavailability either fails hard (blocking downstream processes) or skips the load silently (producing staleness that goes undetected). Neither is acceptable in a production context.

import json
from pathlib import Path
from datetime import date

CACHE_DIR = Path("data/cache/open_meteo")
CACHE_DIR.mkdir(parents=True, exist_ok=True)

def fetch_weather_with_cache(lat: float, lon: float, target_date: date) -> dict:
    cache_path = CACHE_DIR / f"{lat}_{lon}_{target_date}.json"

    # Serve from cache if available (useful for backfills and outage recovery)
    if cache_path.exists():
        with open(cache_path) as f:
            return json.load(f)

    url = "https://archive-api.open-meteo.com/v1/archive"
    params = {
        "latitude": lat,
        "longitude": lon,
        "start_date": str(target_date),
        "end_date": str(target_date),
        "daily": "temperature_2m_max,precipitation_sum,wind_speed_10m_max",
        "timezone": "UTC",
    }

    try:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        # Cache the successful response
        with open(cache_path, "w") as f:
            json.dump(data, f)

        return data

    except requests.RequestException as e:
        # Fallback: return last available cached day if source is down
        cached_files = sorted(CACHE_DIR.glob(f"{lat}_{lon}_*.json"), reverse=True)
        if cached_files:
            print(f"WARNING: API unavailable ({e}). Serving stale cache: {cached_files[0].name}")
            with open(cached_files[0]) as f:
                return json.load(f)
        raise RuntimeError(f"API unavailable and no cache: {e}") from e

Structuring the integration layer

API integrations benefit from a consistent structure that separates the HTTP mechanics from the business logic. The client module handles authentication, rate limiting, and retries. The extractor module handles pagination and result assembly. The loader module handles destination writes. This separation makes each layer independently testable, and it means a change to the API (a new rate limit, a pagination scheme change) affects only the layer responsible for it.

# api_client.py: HTTP mechanics only
class CoinGeckoClient:
    BASE_URL = "https://api.coingecko.com/api/v3"

    def get(self, endpoint: str, params: dict = None) -> dict:
        # handles auth, retries, rate limit backoff
        ...

# extractor.py: assembles full result sets
def extract_market_data(client: CoinGeckoClient, coin_ids: list[str]) -> list[dict]:
    # handles pagination, logging, validation
    ...

# loader.py: writes to destination
def load_market_data(rows: list[dict], conn) -> None:
    # handles idempotency, schema enforcement
    ...

A pipeline built with this structure can swap the destination (Postgres to BigQuery), the source (CoinGecko free tier to Pro), or the retry strategy without touching the other layers. That modularity is not over-engineering for a production integration: it is the minimum structure that makes the integration maintainable over time.