lesson 152 HTTP & API Clients · advanced

JSON responses and HTTP errors

Most APIs answer in JSON. Read the bytes, check what you got, and json.loads them into dictionaries and lists. When the status is 4xx or 5xx, urllib raises HTTPError instead, and its body usually says what went wrong.

Simulated: this runs against results recorded on real Python.

You'll be able to

Parse a JSON body with json.loads(response.read()) and walk the nested result
Catch urllib.error.HTTPError, read its code and its JSON error body
Decide which failures a function should handle (a 404 for one record) and which it should let through

The idea

JSON is the usual body format: json.loads accepts the bytes directly and gives back dictionaries, lists, strings, numbers, booleans and None (see CSV JSON & JSONL). What's inside follows the API's documentation. The USGS answers in GeoJSON: a dictionary with a features list, one dictionary per event, its details under properties and its position under geometry.

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

example-1.py · Earthquakes from a GeoJSON response

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

url = "https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson&starttime=2025-01-01&endtime=2025-01-02&minmagnitude=5"
saved = {url: (200, {"type": "FeatureCollection", "metadata": {"count": 3}, "features": [
    {"id": "us6000pgsd", "properties": {"mag": 5.5, "place": "153 km S of Isangel, Vanuatu"}, "geometry": {"coordinates": [169.3819, -20.9273, 22]}},
    {"id": "us6000pgv4", "properties": {"mag": 6.1, "place": "South Sandwich Islands region"}, "geometry": {"coordinates": [-26.7059, -56.4095, 70]}},
    {"id": "us6000pgx1", "properties": {"mag": 5.3, "place": "southeast of Easter Island"}, "geometry": {"coordinates": [-100.6636, -36.1244, 10]}},
]})}
opener = build_opener(Replay(saved))

with opener.open(url, timeout=10) as response:
    data = json.loads(response.read())
print(data["metadata"]["count"], "events")
for feature in data["features"]:
    longitude, latitude, depth_km = feature["geometry"]["coordinates"]
    print(feature["properties"]["mag"], feature["properties"]["place"], f"({depth_km} km deep)")

Not every request succeeds. When the status is 4xx or 5xx, open doesn't return a response at all: it raises `urllib.error.HTTPError`. Asking OpenAlex for a work ID that doesn't exist:

example-2.py · A 404 is an exception

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

opener = build_opener(Replay({}))
response = opener.open("https://api.openalex.org/works/W0000000000", timeout=10)
print(response.status)

Which errors to handle depends on the function's job. When the question is *does this record exist?*, a 404 is an answer, so return None. A 400 means your request is wrong and a 5xx means the server failed: those should reach the caller, ideally with the API's own explanation from the error body.

example-3.py · Handle the 404, report the rest

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

def get_json(opener, url):
    try:
        with opener.open(url, timeout=10) as response:
            return json.loads(response.read())
    except HTTPError as err:
        if err.code == 404:
            return None
        details = json.loads(err.read())
        raise RuntimeError(f"{err.code} from the API: {details['message']}") from err

saved = {
    "https://api.openalex.org/works/W3003257820": (200, {"id": "https://openalex.org/W3003257820", "publication_year": 2020, "cited_by_count": 38040}),
    "https://api.openalex.org/works?filter=publication_yr:2020": (400, {"error": "Invalid query parameters error.", "message": "publication_yr is not a valid field."}),
}
opener = build_opener(Replay(saved))

print(get_json(opener, "https://api.openalex.org/works/W3003257820")["cited_by_count"])
print(get_json(opener, "https://api.openalex.org/works/W0000000000"))
try:
    get_json(opener, "https://api.openalex.org/works?filter=publication_yr:2020")
except RuntimeError as err:
    print(err)

When to use it: every JSON API. Mistakes to avoid: expecting open to return a response for a 404 (it raises); catching HTTPError and returning None for every status, which hides a broken request or a failing server behind "no data"; and assuming every field is present: real records leave optional fields out.

Review

json.loads(response.read()) turns a JSON body into Python data; read the API's documentation for its shape.
4xx and 5xx statuses raise HTTPError; err.code and err.read() say what went wrong.
Handle only the errors that answer the function's question; re-raise the rest.
Next: APIs that send large results one page at a time.

Practise this