HTTP and APIs with httpx

What you'll learn

• Understand a request and a response: method, URL, headers, status code, body
• Call a JSON API with httpx: GET with query parameters, POST with a JSON body
• Handle status codes, timeouts, and connection errors without crashing
• Test against a local server so nothing depends on the internet

Concept

Most useful data lives behind an HTTP API: a URL you send a request to, and a response that comes back, usually as JSON. A request has a method that says what you want, GET to read, POST to create, PUT or PATCH to update, DELETE to remove; a URL that says which resource; headers with metadata such as the content type or an authorization token; and, for writes, a body. A response has a status code, headers, and a body. Everything else about web APIs is detail on top of that shape.

httpx is the client this course uses: the same shape as the older requests, with timeouts on by default and an async version for the concurrency lessons. httpx.get(url, params={...}) builds the query string; httpx.post(url, json={...}) sends a JSON body with the right content type. The response has .status_code, .headers, .text, and .json() to parse the body. response.raise_for_status() turns any 4xx or 5xx into an HTTPStatusError, which is the simplest way to stop bad responses from being treated as data. For several requests to one host, httpx.Client() reuses the connection and holds shared headers such as the auth token.

Networks fail, so the client must expect it. Always set a timeout; httpx defaults to five seconds and you should choose deliberately. Catch httpx.TimeoutException and httpx.ConnectError for the network layer, and httpx.HTTPStatusError after raise_for_status() for the application layer, and decide per case whether to retry, fall back, or report. The retry helper from the error-handling lesson applies directly: retry timeouts and 5xx, never 4xx.

Code that calls a real API cannot be tested against that API: it is slow, it changes, it needs credentials, and it may not be reachable. The examples here run a tiny server from the standard library's http.server in a background thread on localhost, so every request is real HTTP but nothing leaves the machine and the output is fixed. In tests you do the same, or you pass httpx.MockTransport to a client so responses are produced by a function you wrote, which is the client's own way of being faked.

Code

A tiny JSON API served from a background thread. Every later block imports start_server from this file; it is written once and not run on its own.

# server.py
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

USERS = {1: {"id": 1, "name": "ada", "active": True}, 2: {"id": 2, "name": "linus", "active": False}}


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def send_json(self, status, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        url = urlparse(self.path)
        if url.path == "/users":
            active = parse_qs(url.query).get("active")
            users = list(USERS.values())
            if active:
                users = [u for u in users if u["active"] == (active[0] == "true")]
            return self.send_json(200, users)
        if url.path.startswith("/users/"):
            user = USERS.get(int(url.path.rsplit("/", 1)[1]))
            return self.send_json(200, user) if user else self.send_json(404, {"error": "no such user"})
        if url.path == "/delay":
            time.sleep(0.2)
            return self.send_json(200, {"waited": 0.2})
        if url.path == "/slow":
            time.sleep(2)
            return self.send_json(200, {})
        if url.path == "/boom":
            return self.send_json(503, {"error": "try later"})
        self.send_json(404, {"error": "unknown path"})

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        data = json.loads(self.rfile.read(length) or b"{}")
        if "name" not in data:
            return self.send_json(400, {"error": "name is required"})
        new_id = max(USERS) + 1
        USERS[new_id] = {"id": new_id, "name": data["name"], "active": True}
        self.send_json(201, USERS[new_id])


class Server(ThreadingHTTPServer):
    request_queue_size = 32          # ten connections arriving at once must not overflow the backlog


def start_server():
    server = Server(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return f"http://127.0.0.1:{server.server_port}"

GET with query parameters, a single resource, and a POST with a JSON body. Status codes, headers, and parsed JSON come off the response object.

# requests_basic.py
import httpx

from server import start_server

base = start_server()

r = httpx.get(f"{base}/users", params={"active": "true"}, timeout=5)
print(r.status_code, r.headers["content-type"])
print(r.json())

r = httpx.get(f"{base}/users/2", timeout=5)
print(r.status_code, r.json()["name"])

r = httpx.post(f"{base}/users", json={"name": "grace"}, timeout=5)
print(r.status_code, r.json())
print(len(httpx.get(f"{base}/users", timeout=5).json()), "users now")

200 application/json
[{'id': 1, 'name': 'ada', 'active': True}]
200 linus
201 {'id': 3, 'name': 'grace', 'active': True}
3 users now

Error handling at both layers: a 404 that is expected, a 400 and a 503 caught through raise_for_status, a timeout, and a refused connection.

# errors.py
import httpx

from server import start_server

base = start_server()

r = httpx.get(f"{base}/users/99", timeout=5)
print(r.status_code, r.json()["error"])

for path, kwargs in [("/users", {"json": {}}), ("/boom", {})]:
    try:
        method = httpx.post if kwargs else httpx.get
        method(f"{base}{path}", timeout=5, **kwargs).raise_for_status()
    except httpx.HTTPStatusError as err:
        print("status error:", err.response.status_code, err.response.json()["error"])

try:
    httpx.get(f"{base}/slow", timeout=0.2)
except httpx.TimeoutException:
    print("timed out after 0.2s")

try:
    httpx.get("http://127.0.0.1:9", timeout=1)
except httpx.ConnectError:
    print("connection refused")

404 no such user
status error: 400 name is required
status error: 503 try later
timed out after 0.2s
connection refused

A client with shared settings, and the code you would actually ship: a function that takes a client, so tests can hand it a fake one. The MockTransport version needs no server at all.

# client_and_mock.py
import httpx

from server import start_server


def fetch_active_names(client):
    response = client.get("/users", params={"active": "true"})
    response.raise_for_status()
    return [u["name"] for u in response.json()]


with httpx.Client(base_url=start_server(), timeout=5, headers={"Authorization": "Bearer demo-token"}) as client:
    print("real server:", fetch_active_names(client))
    print("auth header sent:", client.headers["authorization"])


def fake_api(request):
    assert request.url.path == "/users"
    return httpx.Response(200, json=[{"name": "test-user", "active": True}])


with httpx.Client(base_url="http://fake", transport=httpx.MockTransport(fake_api)) as client:
    print("mock transport:", fetch_active_names(client))

real server: ['ada']
auth header sent: Bearer demo-token
mock transport: ['test-user']

Retrying transient failures with the pattern from the error-handling lesson: 5xx and timeouts are retried with backoff; a 4xx is not.

# retry_http.py
import httpx

from server import start_server

base = start_server()
attempts = {"n": 0}


def get_with_retry(client, path, tries=3):
    last = None
    for attempt in range(1, tries + 1):
        try:
            response = client.get(path)
            if response.status_code >= 500:
                raise httpx.HTTPStatusError("server error", request=response.request, response=response)
            return response
        except (httpx.TimeoutException, httpx.HTTPStatusError) as err:
            attempts["n"] += 1
            last = err
            print(f"  attempt {attempt}: {type(err).__name__}")
    raise RuntimeError(f"gave up on {path}") from last


with httpx.Client(base_url=base, timeout=5) as client:
    print(get_with_retry(client, "/users/1").json()["name"])
    try:
        get_with_retry(client, "/boom")
    except RuntimeError as err:
        print(err, "| cause:", type(err.__cause__).__name__)
    r = client.get("/users/99")
    print("4xx not retried:", r.status_code, attempts["n"], "retry attempts total")

ada
  attempt 1: HTTPStatusError
  attempt 2: HTTPStatusError
  attempt 3: HTTPStatusError
gave up on /boom | cause: HTTPStatusError
4xx not retried: 404 3 retry attempts total

Try it yourself

Extend server.py with a DELETE /users/<id> route and a PATCH that toggles active. Write a UserApi class that wraps an httpx.Client with methods list, get, create, deactivate, and delete, each calling raise_for_status. Test it twice: once against the local server, and once with MockTransport handlers that return canned responses including a 404 and a 503, asserting your class raises the right errors.

Common mistakes

• No timeout. A request to a server that never answers hangs the program forever; httpx defaults to five seconds, and you should set it explicitly for every call.
• Treating any response as data. A 500 with an HTML error page has a .text too; call raise_for_status() or check the code before .json().
• Testing against the real API. It is slow, flaky, rate-limited, and may cost money; use a local server or MockTransport, and hit the real thing only in a separate, optional integration test.

What's next

• A request is method, URL, headers, and body; a response is status, headers, and body; JSON in and out via params=, json=, and .json().
• Set timeouts, raise_for_status(), and catch HTTPStatusError, TimeoutException, and ConnectError deliberately; retry only transient failures.
• Take the client as a parameter, and test with a local http.server or httpx.MockTransport.

Your programs can talk to the web. Next you will learn the patterns that shape how larger programs are organized, translated into Python's plain functions, classes, and dictionaries.