API Reliability and Production Patterns

13.2. API Reliability and Production Patterns#

Learning goals — By the end of this section you will be able to:

  • Use authentication headers and environment variables safely

  • Handle paginated APIs with a loop that accumulates results

  • Implement retry logic with exponential backoff for transient failures

  • Validate JSON response contracts defensively before downstream processing

import os
import time
import random
import requests

13.2.1. POST Requests#

A GET request retrieves data from a server. A POST request sends data to a server — typically to create a resource or submit a form.

response = requests.post('https://httpbin.org/post', json={'name': 'Alice', 'role': 'student'})
response.raise_for_status()
data = response.json()
print('Status:', response.status_code)
print('Server received JSON:', data['json'])

Status: 200
Server received JSON: {'name': 'Alice', 'role': 'student'}

13.2.2. Headers and API Keys#

Use headers={} for auth metadata. Never hard-code real keys in source files. Load secrets from environment variables (for example, os.environ).

api_key = os.environ.get('DEMO_API_KEY', 'demo-key')
headers = {'Authorization': f'Bearer {api_key}', 'Accept': 'application/json'}
resp = requests.get('https://httpbin.org/headers', headers=headers, timeout=10)
resp.raise_for_status()
print(resp.json()['headers'])

{'Accept': 'application/json', 'Accept-Encoding': 'gzip, deflate', 'Authorization': 'Bearer demo-key', 'Host': 'httpbin.org', 'User-Agent': 'python-requests/2.32.5', 'X-Amzn-Trace-Id': 'Root=1-6a04f0ef-75e702274a11c013518af47c'}

13.2.3. Pagination Patterns#

Most APIs return data in chunks. Loop until there is no next page token.

# Mock paginated API responses for teaching
pages = {
    1: {'items': ['a', 'b'], 'next_page': 2},
    2: {'items': ['c', 'd'], 'next_page': 3},
    3: {'items': ['e'], 'next_page': None},
}

all_items = []
page = 1
while page is not None:
    payload = pages[page]
    all_items.extend(payload['items'])
    page = payload['next_page']

print('Collected:', all_items)

Collected: ['a', 'b', 'c', 'd', 'e']

13.2.4. Retry with Exponential Backoff#

Retry transient failures (timeouts, 429, 5xx) with increasing wait times.

def get_with_retry(url, params=None, retries=3, timeout=5):
    delay = 0.5
    for attempt in range(1, retries + 1):
        try:
            resp = requests.get(url, params=params, timeout=timeout)
            if resp.status_code in (429, 500, 502, 503, 504):
                raise requests.HTTPError(f'Transient HTTP {resp.status_code}', response=resp)
            resp.raise_for_status()
            return resp
        except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as e:
            if attempt == retries:
                raise
            sleep_for = delay * (2 ** (attempt - 1)) + random.uniform(0, 0.2)
            print(f'Retry {attempt}/{retries - 1} after {sleep_for:.2f}s -> {type(e).__name__}')
            time.sleep(sleep_for)

13.2.5. Response Contract Validation#

Validate required keys before downstream processing.

def parse_weather_response(payload):
    if 'current_weather' not in payload:
        raise ValueError('Missing current_weather in API response')

    current = payload['current_weather']
    if 'temperature' not in current or 'windspeed' not in current:
        raise ValueError('Missing required weather fields: temperature/windspeed')

    return {'temperature': current['temperature'], 'windspeed': current['windspeed']}

sample = {'current_weather': {'temperature': 21.3, 'windspeed': 5.8}}
print(parse_weather_response(sample))

{'temperature': 21.3, 'windspeed': 5.8}

### Exercise: Build a Resilient API Client
# 1) Write fetch_all_pages(get_page_fn) that keeps requesting until no next page
# 2) Wrap calls with retry for Timeout and transient HTTP errors
# 3) Validate each page has an items key before extending results

# Your code here

Hide code cell source

 
### Solution
def fetch_all_pages(get_page_fn, retries=3):
    items = []
    page = 1

    while page is not None:
        delay = 0.5
        for attempt in range(1, retries + 1):
            try:
                payload = get_page_fn(page)
                if 'items' not in payload:
                    raise ValueError('Page payload missing items key')
                items.extend(payload['items'])
                page = payload.get('next_page')
                break
            except Exception:
                if attempt == retries:
                    raise
                time.sleep(delay * (2 ** (attempt - 1)))

    return items

mock_pages = {1: {'items': [10, 20], 'next_page': 2}, 2: {'items': [30], 'next_page': None}}

def mock_get_page(p):
    return mock_pages[p]

print(fetch_all_pages(mock_get_page))

[10, 20, 30]

13.2.6. Summary#

Concept

Key idea

Auth headers + env vars

Keep secrets out of source code; load from environment

Pagination loop

Request pages until no next page/cursor remains

Retry + backoff

Retry transient failures with increasing wait times

Defensive validation

Check required JSON keys before business logic
Contents