Module 4: Consuming Public APIs
Error handling and formatted output
Capsule overview
Your CLI works. It consumes 5 APIs, extracts data, and prints it in the terminal. But if the network goes down, you get a Python traceback. If an API returns 429 (rate limit), the user sees "HTTP 429" with no context. If someone runs python cli.py with no arguments, they have no idea what to do. A professional CLI doesn't behave like that.
This capsule turns your CLI from "working" into "production-ready". You'll build a unified error handling layer that catches every possible error — connection errors, timeouts, HTTP 4xx/5xx, malformed JSON — and translates them into messages a human understands. You'll add automatic retry for transient errors (429 and 5xx). And you'll create an output system with readable tables, a JSON option via --json, and debugging via --verbose.
Classifying API errors
When you consume an API, there are exactly 5 categories of errors. Your code has to handle all of them:
Category │ Example │ Retry?
────────────────────┼────────────────────────────────┼──────────
CONNECTION │ DNS fails, server unreachable │ Yes (1-2)
TIMEOUT │ Server doesn't respond in Ns │ Yes (1-2)
HTTP_CLIENT (4xx) │ 400 Bad Request, 404 Not Found │ No
HTTP_SERVER (5xx) │ 500 Internal Server Error │ Yes (2-3)
RATE_LIMIT (429) │ Too Many Requests │ Yes (wait)
JSON_PARSE │ Response isn't valid JSON │ No
Connection and timeout errors are on the client side (your network). 4xx errors are your fault (a malformed URL, an invalid API key). 5xx errors are the server's fault. 429s are rate limiting.
Friendly error messages
The user should never see a Python traceback. Every error translates into an action:
import json
def friendly_error(error_type, status=None, api_name="API", detail=None):
"""Translates a technical error into a friendly message."""
messages = {
"CONNECTION": (f"Could not connect to {api_name}. "
"Check your internet connection."),
"TIMEOUT": (f"{api_name} didn't respond in time. "
"Try again in a few seconds."),
"RATE_LIMIT": (f"Too many requests to {api_name}. "
"Wait a moment before trying again."),
"JSON_PARSE": (f"{api_name} returned an unexpected response."),
"CONFIG": (f"Missing configuration for {api_name}. "
"Check that the API keys are in your .env file"),
}
if error_type in messages:
return messages[error_type]
if status:
status_messages = {
400: "The request has an error. Check the parameters.",
401: f"You aren't authorized. Check your API key.",
403: f"Access forbidden to {api_name}.",
404: f"The resource was not found in {api_name}.",
500: f"{api_name} had an internal error. Try again later.",
503: f"{api_name} is unavailable. Try again later.",
}
msg = status_messages.get(status, f"HTTP error {status}.")
if detail:
msg += f" ({detail})"
return msg
return f"Unknown error connecting to {api_name}."
test_cases = [
("CONNECTION", None, "GitHub", None),
("TIMEOUT", None, "OpenWeather", None),
("RATE_LIMIT", 429, "GitHub", None),
("HTTP_ERROR", 404, "REST Countries", "Not Found"),
("HTTP_ERROR", 401, "OpenWeather", "Invalid API key"),
("CONFIG", None, "OpenWeather", None),
]
for error_type, status, api, detail in test_cases:
msg = friendly_error(error_type, status, api, detail)
status_str = str(status) if status else "N/A"
print(f" [{status_str:>3}] {msg}")
Expected output:
[N/A] Could not connect to GitHub. Check your internet connection.
[N/A] OpenWeather didn't respond in time. Try again in a few seconds.
[429] Too many requests to GitHub. Wait a moment before trying again.
[404] The resource was not found in REST Countries. (Not Found)
[401] You aren't authorized. Check your API key. (Invalid API key)
[N/A] Missing configuration for OpenWeather. Check that the API keys are in your .env file
Every message says what happened and what to do. It doesn't say "HTTP 401 Unauthorized" — it says "Check your API key."
Retry logic with exponential backoff
When an API returns 429 or 5xx, you don't retry immediately — that makes the problem worse. Use exponential backoff: wait 1s, then 2s, then 4s:
import requests
import time
def request_with_retry(url, max_retries=3, timeout=10):
"""GET with automatic retry for transient errors."""
for attempt in range(max_retries + 1):
try:
response = requests.get(url, timeout=timeout)
except requests.exceptions.ConnectionError:
if attempt < max_retries:
wait = 2 ** attempt
print(f" ⏳ Connection failed. Retrying in {wait}s... "
f"({attempt + 1}/{max_retries})")
time.sleep(wait)
continue
return {"success": False, "error": "Connection failed after retries"}
except requests.exceptions.Timeout:
if attempt < max_retries:
wait = 2 ** attempt
print(f" ⏳ Timeout. Retrying in {wait}s... "
f"({attempt + 1}/{max_retries})")
time.sleep(wait)
continue
return {"success": False, "error": "Timeout after retries"}
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e)}
if response.status_code == 429 or response.status_code >= 500:
if attempt < max_retries:
wait = 2 ** attempt
if response.status_code == 429:
retry_after = response.headers.get("Retry-After")
if retry_after and retry_after.isdigit():
wait = int(retry_after)
print(f" ⏳ HTTP {response.status_code}. "
f"Retrying in {wait}s... ({attempt + 1}/{max_retries})")
time.sleep(wait)
continue
data = None
try:
data = response.json()
except Exception:
pass
if not response.ok:
return {"success": False, "error": f"HTTP {response.status_code}"}
return {"success": True, "data": data}
return {"success": False, "error": "Max retries exceeded"}
result = request_with_retry("https://api.github.com/users/octocat")
if result["success"]:
print(f" ✅ User: {result['data']['login']}")
result = request_with_retry("https://api.github.com/users/does-not-exist-xyz-99999")
print(f" {'✅' if result['success'] else '❌'} {result.get('error', 'OK')}")
Expected output:
✅ User: octocat
❌ HTTP 404
The 404 isn't retried — it's a client error. Only 429s and 5xx trigger the retry. The Retry-After header (when present) overrides the calculated backoff.
Formatted output: tables and key-value
import json
def format_table(headers, rows, indent=2):
"""Formats data as a table with aligned columns."""
col_widths = []
for i, header in enumerate(headers):
max_width = len(header)
for row in rows:
if i < len(row):
max_width = max(max_width, len(str(row[i])))
col_widths.append(min(max_width, 40))
prefix = " " * indent
header_line = prefix + " ".join(h.ljust(w) for h, w in zip(headers, col_widths))
separator = prefix + " ".join("─" * w for w in col_widths)
lines = [header_line, separator]
for row in rows:
cells = []
for cell, width in zip(row, col_widths):
s = str(cell)
if len(s) > width:
s = s[:width - 3] + "..."
cells.append(s.ljust(width))
lines.append(prefix + " ".join(cells))
return "\n".join(lines)
def format_key_value(data, indent=2):
"""Formats a dict as aligned key: value pairs."""
if not data:
return ""
prefix = " " * indent
max_key = max(len(str(k)) for k in data.keys())
lines = []
for key, value in data.items():
if isinstance(value, list):
value = ", ".join(str(v) for v in value)
lines.append(f"{prefix}{str(key).ljust(max_key)} {value}")
return "\n".join(lines)
print("=== Table format ===\n")
print(format_table(
["Country", "Capital", "Population"],
[["Mexico", "Mexico City", "128,932,753"],
["Japan", "Tokyo", "125,836,021"],
["Brazil", "Brasília", "212,559,417"]]
))
print("\n\n=== Key-value format ===\n")
print(format_key_value({
"City": "London",
"Country": "GB",
"Temperature": "12.5°C",
"Humidity": "72%",
"Description": "scattered clouds",
}))
Expected output:
=== Table format ===
Country Capital Population
─────── ──────────── ─────────────
Mexico Mexico City 128,932,753
Japan Tokyo 125,836,021
Brazil Brasília 212,559,417
=== Key-value format ===
City London
Country GB
Temperature 12.5°C
Humidity 72%
Description scattered clouds
The --json flag for raw output
The --json flag switches the output from readable format to JSON. It's useful for pipes: python cli.py weather London --json | jq .temp.
import json
def output_result(data, json_mode=False, title=None):
"""Shows the result in readable format or as JSON."""
if json_mode:
print(json.dumps(data, indent=2, ensure_ascii=False))
return
if title:
print(f"\n {title}")
print(f" {'─' * len(title)}")
if isinstance(data, dict):
max_key = max(len(str(k)) for k in data.keys())
for key, value in data.items():
if isinstance(value, list):
value = ", ".join(str(v) for v in value)
print(f" {str(key).ljust(max_key)} {value}")
print()
def output_error(error_msg, json_mode=False):
"""Shows the error in readable format or as JSON."""
if json_mode:
print(json.dumps({"error": error_msg}, indent=2, ensure_ascii=False))
else:
print(f"\n ❌ {error_msg}\n")
weather_data = {
"City": "London, GB",
"Temperature": "12.5°C",
"Humidity": "72%",
"Description": "overcast clouds",
}
print("=== Readable mode ===")
output_result(weather_data, json_mode=False, title="Current weather")
print("=== JSON mode ===")
output_result(weather_data, json_mode=True)
print("\n=== Readable error ===")
output_error("City not found")
print("=== JSON error ===")
output_error("City not found", json_mode=True)
Expected output:
=== Readable mode ===
Current weather
───────────────
City London, GB
Temperature 12.5°C
Humidity 72%
Description overcast clouds
=== JSON mode ===
{
"City": "London, GB",
"Temperature": "12.5°C",
"Humidity": "72%",
"Description": "overcast clouds"
}
=== Readable error ===
❌ City not found
=== JSON error ===
{
"error": "City not found"
}
The --verbose flag for debugging
--verbose shows technical information that's invaluable for diagnosing problems:
import requests
import time
import sys
class VerboseAPIClient:
"""HTTP client with optional detailed logging."""
def __init__(self, base_url, verbose=False, timeout=10):
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.verbose = verbose
self.session = requests.Session()
self.session.headers.update({"Accept": "application/json",
"User-Agent": "RESTClientCLI/1.0"})
def _log(self, message):
if self.verbose:
print(f" 🔍 {message}", file=sys.stderr)
def get(self, path, **kwargs):
url = f"{self.base_url}/{path.lstrip('/')}"
kwargs.setdefault("timeout", self.timeout)
self._log(f"GET {url}")
if "params" in kwargs:
safe_params = {k: (v if k != "appid" else "***")
for k, v in kwargs["params"].items()}
self._log(f"Params: {safe_params}")
start = time.time()
try:
response = self.session.request("GET", url, **kwargs)
elapsed = (time.time() - start) * 1000
except requests.exceptions.RequestException as e:
self._log(f"FAILED: {type(e).__name__}")
return {"success": False, "data": None, "error": str(e)}
self._log(f"Response: {response.status_code} in {elapsed:.0f}ms")
self._log(f"Content-Type: {response.headers.get('Content-Type', 'N/A')}")
data = None
ct = response.headers.get("Content-Type", "")
if "application/json" in ct and response.text:
try:
data = response.json()
except Exception:
return {"success": False, "data": None, "error": "Malformed JSON"}
if not response.ok:
return {"success": False, "data": data,
"error": f"HTTP {response.status_code}"}
return {"success": True, "data": data, "error": None}
def close(self):
self.session.close()
print("=== Without verbose ===\n")
client = VerboseAPIClient("https://api.github.com", verbose=False)
result = client.get("/users/octocat")
if result["success"]:
print(f" User: {result['data']['login']}")
client.close()
print("\n=== With verbose ===\n")
client = VerboseAPIClient("https://api.github.com", verbose=True)
result = client.get("/users/octocat")
if result["success"]:
print(f" User: {result['data']['login']}")
client.close()
Expected output:
=== Without verbose ===
User: octocat
=== With verbose ===
🔍 GET https://api.github.com/users/octocat
🔍 Response: 200 in 234ms
🔍 Content-Type: application/json; charset=utf-8
User: octocat
Notice: OpenWeather's API key gets masked as *** in verbose mode. Never print credentials in logs.
Help text and usage examples
A professional CLI has clear help text:
import argparse
def create_cli():
parser = argparse.ArgumentParser(
prog="cli.py",
description="REST Client CLI — Query multiple APIs from your terminal",
epilog="""
Usage examples:
python cli.py weather London
python cli.py weather "Mexico City" --json
python cli.py countries Mexico
python cli.py github user octocat
python cli.py dogs random 3
Configuration:
Create a .env file with your API keys:
OPENWEATHER_API_KEY=your_key_here
GITHUB_TOKEN=your_token_here (optional)
""",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--json", action="store_true",
help="Output in JSON format")
parser.add_argument("--verbose", action="store_true",
help="Show debugging details")
subparsers = parser.add_subparsers(dest="command", help="API to query")
wp = subparsers.add_parser("weather", help="Weather (OpenWeather)")
wp.add_argument("city", help="City (e.g. London, 'New York')")
cp = subparsers.add_parser("countries", help="Countries (REST Countries)")
cp.add_argument("query", help="Name or ISO code (e.g. Mexico, JP)")
gp = subparsers.add_parser("github", help="GitHub API")
gp.add_argument("action", choices=["user", "repos"])
gp.add_argument("target", help="GitHub username")
return parser
parser = create_cli()
parser.parse_args(["--help"])
The epilog with RawDescriptionHelpFormatter preserves the formatting. The examples tell the user exactly what to type.
Troubleshooting
Problem 1: The retry runs forever
Cause: max_retries isn't set, or the loop has no exit condition.
Solution: Always define max_retries (2-3). The loop uses range(max_retries + 1) and exits with break when the error isn't retryable.
Problem 2: --json and --verbose produce mixed output
Cause: The verbose messages get mixed into the JSON.
Solution: When --json is active, send verbose output to stderr:
import sys
def _log(self, message):
if self.verbose:
print(f" 🔍 {message}", file=sys.stderr)
Problem 3: JSON errors aren't being caught
Cause: You're using response.json() without a try/except.
Solution: Always wrap response.json() in try/except and check the Content-Type before parsing.
Exercises
Exercise 1: Unified error handler (Easy)
Create a handle_api_result(result, api_name, json_mode=False) function that takes the result dict from any API client and shows either the formatted result or the friendly error.
See solution
import json
def friendly_error(error_type, status=None, api_name="API"):
messages = {
"CONNECTION": f"Could not connect to {api_name}.",
"TIMEOUT": f"{api_name} didn't respond in time.",
"RATE_LIMIT": f"Too many requests to {api_name}.",
"CONFIG": f"Missing configuration for {api_name}.",
}
if error_type in messages:
return messages[error_type]
if status == 404:
return f"Resource not found in {api_name}."
if status and 400 <= status < 500:
return f"Request error to {api_name} (HTTP {status})."
if status and 500 <= status < 600:
return f"Server error in {api_name} (HTTP {status})."
return f"Unknown error in {api_name}."
def handle_api_result(result, api_name, json_mode=False):
if result["success"]:
if json_mode:
print(json.dumps(result["data"], indent=2, ensure_ascii=False))
else:
print(f"\n ✅ {api_name}:")
data = result["data"]
if isinstance(data, dict):
mx = max(len(str(k)) for k in data.keys())
for k, v in data.items():
print(f" {str(k).ljust(mx)} {v}")
print()
else:
error_type = result.get("error_type", "UNKNOWN")
msg = friendly_error(error_type, result.get("status"), api_name)
if json_mode:
print(json.dumps({"error": msg}, indent=2))
else:
print(f"\n ❌ {msg}\n")
handle_api_result(
{"success": True, "data": {"city": "London", "temp": "12.5°C"}},
"OpenWeather")
handle_api_result(
{"success": False, "error": "city not found", "error_type": "HTTP_CLIENT",
"status": 404},
"OpenWeather")
Explanation: handle_api_result() is the single point where every result turns into output.
Exercise 2: Request logger (Medium)
Create a RequestLogger class that records each request (URL, method, status, time, success) and has a summary() method with statistics.
See solution
import time
class RequestLogger:
def __init__(self):
self._entries = []
def log(self, method, url, status, elapsed_ms, success):
self._entries.append({
"method": method, "url": url, "status": status,
"ms": elapsed_ms, "success": success,
"timestamp": time.strftime("%H:%M:%S"),
})
def summary(self):
total = len(self._entries)
if total == 0:
print(" No requests were recorded.")
return
ok = sum(1 for e in self._entries if e["success"])
avg_ms = sum(e["ms"] for e in self._entries) / total
print(f"\n {'═' * 55}")
print(f" 📊 Request summary")
print(f" {'═' * 55}")
print(f" Total: {total} | Successful: {ok} ✅ | "
f"Failed: {total - ok} ❌ | Average: {avg_ms:.0f}ms")
print(f"\n {'Time':<10} {'Method':<7} {'Status':>6} "
f"{'Elapsed':>8} {'URL'}")
print(f" {'─'*10} {'─'*7} {'─'*6} {'─'*8} {'─'*25}")
for e in self._entries:
icon = "✅" if e["success"] else "❌"
st = str(e["status"]) if e["status"] else "ERR"
print(f" {e['timestamp']:<10} {e['method']:<7} {st:>6} "
f"{e['ms']:>6.0f}ms {icon} {e['url']}")
print()
logger = RequestLogger()
logger.log("GET", "/users/octocat", 200, 234, True)
logger.log("GET", "/users/xyz999", 404, 180, False)
logger.log("GET", "/repos", 200, 412, True)
logger.summary()
Explanation: RequestLogger accumulates one dict per request. summary() prints a recap with statistics.
Exercise 3: Output with auto-formatting (Medium)
Create an OutputFormatter class with an auto(data, title) method that detects the data type (list → table, dict → key-value) and formats it automatically, with support for json_mode.
See solution
import json
class OutputFormatter:
def __init__(self, json_mode=False):
self.json_mode = json_mode
def auto(self, data, title=None):
if self.json_mode:
print(json.dumps(data, indent=2, ensure_ascii=False))
return
if isinstance(data, list) and data and isinstance(data[0], dict):
headers = list(data[0].keys())
if title:
print(f"\n {title}\n")
widths = [max(len(h), max(len(str(item.get(h, "")))
for item in data)) for h in headers]
print(" " + " ".join(h.ljust(w) for h, w in zip(headers, widths)))
print(" " + " ".join("─" * w for w in widths))
for item in data:
print(" " + " ".join(
str(item.get(h, "")).ljust(w)
for h, w in zip(headers, widths)))
elif isinstance(data, dict):
if title:
print(f"\n {title}")
print(f" {'─' * len(title)}")
mx = max(len(str(k)) for k in data.keys())
for k, v in data.items():
print(f" {str(k).ljust(mx)} {v}")
print()
fmt = OutputFormatter(json_mode=False)
fmt.auto([{"Country": "Mexico", "Capital": "Mexico City"},
{"Country": "Japan", "Capital": "Tokyo"}], "Countries")
fmt.auto({"City": "London", "Temp": "12.5°C"}, "Weather")
fmt_json = OutputFormatter(json_mode=True)
fmt_json.auto({"City": "London", "Temp": "12.5°C"})
Explanation: auto() detects the type and picks a table or key-value layout. With json_mode=True, everything comes out as JSON.
Exercise 4: Retry with a callback (Hard)
Extend the retry logic to accept an on_retry callback that runs before each retry with info about the error.
See solution
import requests
import time
def default_on_retry(attempt, wait, error_info):
print(f" ⏳ Retry {attempt}: waiting {wait}s "
f"(reason: {error_info.get('reason', 'unknown')})")
def request_with_retry(url, max_retries=3, timeout=10, on_retry=None):
if on_retry is None:
on_retry = default_on_retry
for attempt in range(max_retries + 1):
try:
response = requests.get(url, timeout=timeout)
except requests.exceptions.ConnectionError:
if attempt < max_retries:
wait = 2 ** attempt
on_retry(attempt + 1, wait, {"reason": "connection_error"})
time.sleep(wait)
continue
return {"success": False, "error": "Connection failed"}
except requests.exceptions.Timeout:
if attempt < max_retries:
wait = 2 ** attempt
on_retry(attempt + 1, wait, {"reason": "timeout"})
time.sleep(wait)
continue
return {"success": False, "error": "Timeout"}
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e)}
if (response.status_code == 429 or response.status_code >= 500) \
and attempt < max_retries:
wait = 2 ** attempt
on_retry(attempt + 1, wait, {"reason": f"http_{response.status_code}"})
time.sleep(wait)
continue
if not response.ok:
return {"success": False, "error": f"HTTP {response.status_code}"}
return {"success": True, "data": response.json()}
return {"success": False, "error": "Max retries exceeded"}
retry_log = []
def logging_callback(attempt, wait, error_info):
retry_log.append({"attempt": attempt, **error_info})
print(f" 📝 Logged retry #{attempt}: {error_info['reason']}")
result = request_with_retry("https://api.github.com/users/octocat",
max_retries=2, on_retry=logging_callback)
if result["success"]:
print(f"\n ✅ User: {result['data']['login']}")
print(f" 📊 Retries: {len(retry_log)}")
Explanation: The callback decouples the retry logic from the notification logic. You can pass any function: logging to a file, metrics, whatever.
Summary
- 5 error categories: CONNECTION, TIMEOUT, HTTP_CLIENT (4xx), HTTP_SERVER (5xx), RATE_LIMIT (429), JSON_PARSE — handle them all
- Friendly messages: never show tracebacks — translate every error into an action
- Retry with exponential backoff: wait 1s, 2s, 4s — only for transient errors
- The Retry-After header: when it's there, use it instead of the calculated backoff
- The
--jsonflag: raw JSON for pipes (cli.py weather London --json | jq .temp) - The
--verboseflag: URL, status, timing, Content-Type — never credentials format_table()andformat_key_value(): reusable functions with dynamic widths- Verbose to stderr:
print(..., file=sys.stderr)so you don't pollute the JSON output
Next capsule: Final project: REST Client CLI — you assemble everything into a complete, portfolio-worthy CLI.
Additional resources
- Requests: Errors and Exceptions — The requests exception hierarchy
- HTTP 429 Too Many Requests (MDN) — Spec for status 429 and Retry-After
- Exponential Backoff and Jitter (AWS) — Retry patterns
- Python argparse: RawDescriptionHelpFormatter — Formatting help text
- 12 Factor App: Config — Why credentials belong in env vars
- Click Library — An alternative to argparse for more complex CLIs