12.2. API Reliability and Production Patterns#
API code becomes production code when other code depends on it. Reliability means the client handles secrets, pagination, retries, and response validation deliberately instead of assuming every request succeeds and every response has the expected shape.
12.2.1. POST Requests#
A GET request retrieves data. A POST request sends data, often to create a record or submit a form. In requests, use json= when the request body should be JSON. The library serializes the dictionary and sets the appropriate content type.
12.2.1.1. Worked Example: Prepare a JSON POST#
This example prepares a POST request without sending it, so you can inspect what json= does: it sets the Content-Type header to application/json and serializes the dictionary into the request body.
payload = {"name": "Alice", "role": "student"}
request = requests.Request(
"POST",
"https://api.example.test/users",
json=payload,
).prepare()
print(request.method)
print(request.headers["Content-Type"])
print(request.body.decode())
POST
application/json
{"name": "Alice", "role": "student"}
12.2.1.2. Checkpoint: POST Payload#
Prepare a POST request to https://api.example.test/events with JSON fields event_type and count. Print the method and encoded body.
### Your code starts here
payload = {}
request = requests.Request("POST", "https://api.example.test/events", json=payload).prepare()
print(request.method)
print(request.body)
### Your code ends here
POST
b'{}'
POST
{"event_type": "login", "count": 3}
12.2.2. Headers and API Keys#
Headers carry request metadata. Authentication tokens often belong in an Authorization header, not in the URL. Some APIs expect the key in a custom header instead, most commonly X-API-Key; the API’s documentation tells you which header name to use. Keep real secrets out of notebooks and source files: read them from environment variables and fail clearly when they are missing.
12.2.2.1. Worked Example: Build Auth Headers#
This function reads the API key from an environment variable and places it in an Authorization header. If the variable is not set, it falls back to a placeholder so the example still runs; in real code, a missing key should stop the program with a clear error.
def build_headers(token_env_var="DEMO_API_KEY"):
token = os.environ.get(token_env_var)
if token is None:
token = "demo-token"
return {
"Authorization": f"Bearer {token}",
"Accept": "application/json",
}
headers = build_headers()
print(headers["Authorization"])
print(headers["Accept"])
Bearer demo-token
application/json
12.2.2.2. Checkpoint: API Key Header#
Some APIs expect the key in a custom header instead of Authorization. Complete build_key_headers() so it reads the key from the environment variable named by env_var, uses fallback when the variable is not set, and returns a dictionary with an X-API-Key header and an Accept header of application/json. The test code attaches the headers to a prepared GET request.
def build_key_headers(env_var="REPORTS_API_KEY", fallback="test-key"):
### Your code starts here
return {}
### Your code ends here
headers = build_key_headers()
request = requests.Request("GET", "https://api.example.test/reports", headers=headers).prepare()
print(request.headers.get("X-API-Key"))
print(request.headers.get("Accept"))
None
None
test-key
application/json
12.2.3. Nested JSON Parsing#
Many API responses contain dictionaries inside lists inside dictionaries. Direct indexing is concise when the response contract is guaranteed. The .get() method is safer when fields may be optional.
12.2.3.1. Worked Example: Direct and Defensive Access#
The same payload is read in two ways. Direct indexing reaches the user’s name, which the response contract guarantees. Chained .get() calls with defaults read the first metric, so a response without a metrics key returns 0 instead of raising KeyError.
payload = {
"user": {
"id": 101,
"profile": {"name": "Nia", "department": "Analytics"},
},
"metrics": [{"name": "requests", "value": 42}],
}
name = payload["user"]["profile"]["name"]
first_metric = payload.get("metrics", [{}])[0].get("value", 0)
print(name)
print(first_metric)
Nia
42
12.2.3.2. Checkpoint: Defensive Parsing#
Write get_department(payload) so it returns the nested department value when present and 'unknown' otherwise.
def get_department(payload):
### Your code starts here
return None
### Your code ends here
print(get_department(payload))
print(get_department({"user": {}}))
None
None
Analytics
unknown
12.2.4. Pagination Patterns#
APIs often return large result sets in pages. A client should keep requesting pages until the response says there is no next page. Page-based APIs often use an integer page number; cursor-based APIs use a token returned by the previous response. With a cursor, each response includes a value such as next_cursor that you send with the next request, and a missing or None cursor means you have reached the last page.
12.2.4.1. Worked Example: Page-Based Pagination#
A small dictionary stands in for an API that returns one page at a time. fetch_all_pages() keeps requesting the next page and collecting its items until a page reports next_page as None.
pages = {
1: {"items": ["a", "b"], "next_page": 2},
2: {"items": ["c", "d"], "next_page": 3},
3: {"items": ["e"], "next_page": None},
}
def get_page(page_number):
return pages[page_number]
def fetch_all_pages(get_page_fn, first_page=1):
items = []
page = first_page
while page is not None:
payload = get_page_fn(page)
items.extend(payload["items"])
page = payload["next_page"]
return items
print(fetch_all_pages(get_page))
['a', 'b', 'c', 'd', 'e']
12.2.4.2. Checkpoint: Cursor-Based Pagination#
The dictionary cursor_pages stands in for a cursor-based API. Each response holds a results list and a next_cursor token that names the next page; the last page sets next_cursor to None. Complete fetch_all_cursor(get_fn) so it starts at the cursor "start", follows each next_cursor token, and returns every result in order.
cursor_pages = {
"start": {"results": [101, 102], "next_cursor": "k7"},
"k7": {"results": [103, 104], "next_cursor": "m2"},
"m2": {"results": [105], "next_cursor": None},
}
def get_by_cursor(cursor):
return cursor_pages[cursor]
def fetch_all_cursor(get_fn, first_cursor="start"):
results = []
### Your code starts here
### Your code ends here
return results
print(fetch_all_cursor(get_by_cursor))
[]
[101, 102, 103, 104, 105]
12.2.5. Retry with Exponential Backoff#
Retries help with transient failures: timeouts, rate limits, and temporary server errors. Exponential backoff increases the delay after each failed attempt. In production, add jitter so many clients do not retry at exactly the same time.
12.2.5.1. Worked Example: Retry a Flaky Function#
FlakySource raises a timeout a set number of times before it succeeds, which mimics a temporary network problem. retry_call() tries again after each failure, doubling the wait each time, and re-raises the error if the last attempt also fails.
class FlakySource:
def __init__(self, failures_before_success):
self.failures_before_success = failures_before_success
self.calls = 0
def get(self):
self.calls += 1
if self.calls <= self.failures_before_success:
raise requests.Timeout("temporary timeout")
return {"status": "ok", "calls": self.calls}
def retry_call(call_fn, retries=3, base_delay=0.01):
for attempt in range(1, retries + 1):
try:
return call_fn()
except requests.Timeout:
if attempt == retries:
raise
delay = base_delay * (2 ** (attempt - 1))
time.sleep(delay)
source = FlakySource(failures_before_success=2)
print(retry_call(source.get, retries=4))
{'status': 'ok', 'calls': 3}
12.2.5.2. Checkpoint: Retry on Server Errors#
A 5xx status code means the server had a temporary problem, so the request is worth retrying. make_fake_call() returns a stand-in function that replies with the given status codes in order. Complete retry_on_server_error(call_fn) so it calls call_fn(), retries with exponential backoff while the status is 500 or higher, and returns a tuple of the final response and the list of delays it waited. After retries attempts, return the last response even if it is still an error.
def make_fake_call(statuses):
replies = iter(statuses)
def call():
return {"status": next(replies)}
return call
def retry_on_server_error(call_fn, retries=4, base_delay=0.001):
delays = []
### Your code starts here
response = call_fn()
### Your code ends here
return response, delays
print(retry_on_server_error(make_fake_call([503, 500, 200])))
print(retry_on_server_error(make_fake_call([404])))
({'status': 503}, [])
({'status': 404}, [])
({'status': 200}, [0.001, 0.002])
({'status': 404}, [])
({'status': 503}, [0.001, 0.002, 0.004])
12.2.6. Response Contract Validation#
A response contract defines the keys and value types your code expects. Validate close to the API boundary so downstream analysis code receives clean, predictable data.
12.2.6.1. Worked Example: Validate Required Keys#
parse_weather_response() checks the response before using it. It raises a ValueError that names exactly which fields are missing, so a bad response fails at the API boundary with a clear message instead of deep inside later analysis code.
def parse_weather_response(payload):
if "current_weather" not in payload:
raise ValueError("Missing current_weather in API response")
current = payload["current_weather"]
required = ["temperature", "windspeed"]
missing = [key for key in required if key not in current]
if missing:
raise ValueError(f"Missing weather fields: {missing}")
return {
"temperature": current["temperature"],
"windspeed": current["windspeed"],
}
print(parse_weather_response({"current_weather": {"temperature": 21.4, "windspeed": 13.2}}))
{'temperature': 21.4, 'windspeed': 13.2}
12.2.6.2. Checkpoint: Resilient Client#
Complete fetch_valid_pages(get_page_fn) so it fetches every page and validates that each page contains an items list and a next_page key.
def fetch_valid_pages(get_page_fn):
items = []
page = 1
while page is not None:
payload = get_page_fn(page)
### Your code starts here
page = None
### Your code ends here
return items
print(fetch_valid_pages(get_page))
[]
['a', 'b', 'c', 'd', 'e']
12.2.7. Summary#
Reliable API clients keep fragile work near the edge of the program. Use json= for JSON request bodies, headers for authentication metadata, environment variables for secrets, pagination loops for multi-page data, retries for transient failures, and response validation before downstream processing. These habits make API code easier to test because parsing, validation, and pagination can be exercised with local fixtures before the code ever touches a live service.