lesson 155 HTTP & API Clients · advanced

An API client class

When a program calls an API from many places, URL building, headers, timeouts and error handling get repeated and drift apart. A client class keeps them in one place, offers methods that say what you want, and takes its opener as a parameter so tests can replace the network.

Simulated: this runs against results recorded on real Python.

You'll be able to

Move the base URL, default parameters, headers and timeout into one _get method
Give callers methods named after their questions, returning plain Python data
Translate HTTP errors into the client's own exceptions, and test the client with an injected replay opener

The idea

After four lessons the pieces are clear: build a URL, add headers, open with a timeout, parse JSON, handle errors, follow pages. Copying them into every function that needs data means a change (a new header, a different timeout) has to be made everywhere.

An API client is a class that owns those decisions. Its constructor receives the base URL, the settings, and the opener (dependency injection, the Advanced OOP & Design module): build_opener() in production, a replay opener in a test. One private method, _get(path, params), does the request; public methods such as count(...) or work(...) say what the caller wants and return dictionaries and numbers. Errors become the client's own exception classes, so callers don't need to know urllib.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · A small USGS client

import io
import json
from email.message import Message
from http import HTTPStatus
from urllib.request import BaseHandler, Request, build_opener
from urllib.response import addinfourl

class Replay(BaseHandler):
    """Answers https requests from saved copies of real API responses."""
    handler_order = 100

    def __init__(self, saved):
        self.saved = saved

    def https_open(self, request):
        status, body = self.saved.get(request.full_url, (404, {"error": "Not Found"}))
        headers = Message()
        headers["Content-Type"] = "application/json"
        response = addinfourl(io.BytesIO(json.dumps(body).encode()), headers, request.full_url, status)
        response.msg = HTTPStatus(status).phrase
        return response
from urllib.error import HTTPError
from urllib.parse import urlencode

class ApiError(Exception):
    def __init__(self, status, message):
        super().__init__(f"{status}: {message}")
        self.status = status

class UsgsClient:
    base = "https://earthquake.usgs.gov/fdsnws/event/1/"

    def __init__(self, opener, timeout=10):
        self.opener = opener
        self.timeout = timeout

    def _get(self, path, params):
        url = self.base + path + "?" + urlencode({"format": "geojson", **params})
        request = Request(url, headers={"Accept": "application/json", "User-Agent": "quake-report/1.0"})
        try:
            with self.opener.open(request, timeout=self.timeout) as response:
                return json.loads(response.read())
        except HTTPError as err:
            raise ApiError(err.code, err.reason) from err

    def count(self, start, end, min_magnitude):
        return self._get("count", {"starttime": start, "endtime": end, "minmagnitude": min_magnitude})["count"]

    def places(self, start, end, min_magnitude):
        data = self._get("query", {"starttime": start, "endtime": end, "minmagnitude": min_magnitude})
        return [feature["properties"]["place"] for feature in data["features"]]

base = UsgsClient.base
saved = {
    base + "count?format=geojson&starttime=2025-03-01&endtime=2025-04-01&minmagnitude=7": (200, {"count": 2, "maxAllowed": 20000}),
    base + "query?format=geojson&starttime=2025-03-01&endtime=2025-04-01&minmagnitude=7": (200, {"features": [
        {"properties": {"place": "2025 Mandalay, Burma (Myanmar) Earthquake"}},
        {"properties": {"place": "170 km WSW of Riverton, New Zealand"}},
    ]}),
}
client = UsgsClient(build_opener(Replay(saved)))
print(client.count("2025-03-01", "2025-04-01", 7))
print(client.places("2025-03-01", "2025-04-01", 7))
try:
    client.count("2025-03-01", "2025-04-01", 9)
except ApiError as err:
    print("ApiError", err.status, err)

Because the opener comes in from outside, the client can be tested with no network at all, and the replay's list of requests shows exactly what the client sent. That makes tests like *does count ask for the right URL?* and *does a 503 become an ApiError?* fast and deterministic.

example-2.py · Testing the client's requests

import io
import json
import unittest
from email.message import Message
from http import HTTPStatus
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import BaseHandler, Request, build_opener
from urllib.response import addinfourl

class Spy(BaseHandler):
    """Answers every request with one saved response and remembers what was asked."""
    handler_order = 100

    def __init__(self, status, body):
        self.status, self.body, self.requests = status, body, []

    def https_open(self, request):
        self.requests.append(request)
        headers = Message()
        headers["Content-Type"] = "application/json"
        response = addinfourl(io.BytesIO(json.dumps(self.body).encode()), headers, request.full_url, self.status)
        response.msg = HTTPStatus(self.status).phrase
        return response

class OpenAlexClient:
    def __init__(self, opener, mailto):
        self.opener, self.mailto = opener, mailto

    def cited_by_count(self, work_id):
        url = f"https://api.openalex.org/works/{work_id}?" + urlencode({"mailto": self.mailto})
        with self.opener.open(Request(url), timeout=10) as response:
            return json.loads(response.read())["cited_by_count"]

class ClientTests(unittest.TestCase):
    def test_sends_mailto_and_reads_count(self):
        spy = Spy(200, {"cited_by_count": 8702})
        client = OpenAlexClient(build_opener(spy), "ana@example.com")
        self.assertEqual(client.cited_by_count("W3036167779"), 8702)
        self.assertEqual(spy.requests[0].full_url, "https://api.openalex.org/works/W3036167779?mailto=ana%40example.com")

    def test_server_error_is_raised(self):
        client = OpenAlexClient(build_opener(Spy(503, {"error": "busy"})), "ana@example.com")
        with self.assertRaises(HTTPError):
            client.cited_by_count("W3036167779")

result = unittest.TextTestRunner(stream=io.StringIO()).run(unittest.defaultTestLoader.loadTestsFromTestCase(ClientTests))
print(result.testsRun, "tests,", len(result.failures), "failures,", len(result.errors), "errors")

When to use it: as soon as a program calls an API from more than one place, or needs tests that don't touch the network. Mistakes to avoid: creating the opener inside the class (then nothing can replace it in a test); returning raw response objects instead of data; letting each method build its own URL and headers; and swallowing errors inside the client so callers get None without knowing why.

Review

A client class keeps the base URL, headers, timeout and error handling in one _get method.
Public methods are named after questions and return plain data; errors become the client's own exceptions.
Take the opener as a parameter: build_opener() in production, a replay or spy in tests.
Next (see Iterators & Generators): iterators and generators, which can hand out results like these one at a time.