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 --json flag: raw JSON for pipes (cli.py weather London --json | jq .temp)
  • The --verbose flag: URL, status, timing, Content-Type — never credentials
  • format_table() and format_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

  1. Requests: Errors and Exceptions — The requests exception hierarchy
  2. HTTP 429 Too Many Requests (MDN) — Spec for status 429 and Retry-After
  3. Exponential Backoff and Jitter (AWS) — Retry patterns
  4. Python argparse: RawDescriptionHelpFormatter — Formatting help text
  5. 12 Factor App: Config — Why credentials belong in env vars
  6. Click Library — An alternative to argparse for more complex CLIs