Calling HTTP APIs

Much of modern programming is talking to other services over HTTP: weather data, payments, an LLM, your own company's backend. Making one request is a single line. Making requests that survive slow servers, rate limits, outages, changing response formats and missing API keys, and testing all of that without the network, is what this lesson is about. It uses httpx, a modern HTTP client with both synchronous and async APIs, and Pydantic to validate what comes back.

#A first request

uv add httpx pydantic

import httpx

response = httpx.get(
    "https://api.open-meteo.com/v1/forecast",
    params={"latitude": 52.52, "longitude": 13.41, "daily": "temperature_2m_max", "timezone": "auto"},
)
print(response.status_code)          # 200
print(response.headers["content-type"])   # application/json; charset=utf-8
data = response.json()               # parse the JSON body into dicts and lists
print(data["daily"]["temperature_2m_max"][:3])   # e.g. [14.2, 12.8, 15.1]

params builds the query string for you, with correct escaping, so never glue URLs together with f-strings. For a request body, pass json=... (sets the Content-Type header and encodes it) or data=... for form fields.

#Status codes and errors

A response that arrived isn't necessarily a success. httpx doesn't raise on 4xx or 5xx by default; call raise_for_status():

response = httpx.get("https://api.example.com/users/999")
response.raise_for_status()   # raises httpx.HTTPStatusError for 4xx and 5xx

There are two families of failure, and they need different handling:

How to treat each status class:

• 400, 401, 403, 404, 422: your request is wrong, unauthorised or about something that doesn't exist. Retrying sends the same wrong request; fix the request or report the error.
• 429 Too Many Requests: you're being rate-limited. Wait (the Retry-After header often says how long) and retry.
• 500, 502, 503, 504: the server or something in front of it failed. Often temporary; retry with backoff.

Both exception families inherit from httpx.HTTPError, which is convenient for a final catch-all at the edge of your program.

#Use a Client, and always set timeouts

Top-level httpx.get opens a new connection for every call. A Client keeps connections open and reuses them, which saves a TCP and TLS handshake on every request, and lets you configure everything once:

import os

import httpx

client = httpx.Client(
    base_url="https://api.github.com",
    headers={
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
        "User-Agent": "acme-report/1.0",
    },
    timeout=httpx.Timeout(10.0, connect=3.0),
)

response = client.get("/repos/python/cpython")

Create one client per service at startup, reuse it, and close it on shutdown (or use it as a context manager: with httpx.Client(...) as client:).

Timeouts are not optional. A server that accepts the connection and then never responds will hang your program forever without one. httpx defaults to 5 seconds; set values that fit the API, with a short connect timeout (a server that can't accept a connection in 3 seconds is down) and a longer read timeout for slow endpoints.

#Retries with backoff

Transient failures are normal at scale, and one retry usually succeeds. But retrying immediately, in a tight loop, makes an overloaded server worse. Retry with exponential backoff and jitter: wait longer after each failure, with randomness so many clients don't retry in lockstep.

import random
import time

import httpx

RETRY_STATUSES = {429, 500, 502, 503, 504}


def request_with_retries(
    client: httpx.Client, method: str, url: str, *, attempts: int = 4, **kwargs
) -> httpx.Response:
    for attempt in range(1, attempts + 1):
        try:
            response = client.request(method, url, **kwargs)
            if response.status_code not in RETRY_STATUSES:
                response.raise_for_status()
                return response
            failure: Exception = httpx.HTTPStatusError(
                f"{response.status_code}", request=response.request, response=response
            )
            retry_after = response.headers.get("Retry-After")
        except httpx.TransportError as e:
            failure, retry_after = e, None

        if attempt == attempts:
            raise failure
        if retry_after and retry_after.isdigit():
            delay = float(retry_after)
        else:
            delay = random.uniform(0, 0.5 * 2**attempt)   # full jitter: up to 1s, 2s, 4s...
        time.sleep(delay)
    raise AssertionError("unreachable")

Two rules keep retries safe:

• Only retry what's safe to repeat. GET is fine. A POST that creates a payment, retried after a timeout, may create two payments, because the first may have succeeded before the connection dropped. Retry non-idempotent requests only when the API supports an idempotency key header that lets the server recognise the repeat.
• Cap the total. A few attempts with a maximum delay. Unbounded retries turn a short outage into a pile-up.

Libraries such as tenacity or stamina provide well-tested decorators for this; the function above shows what they do.

#Pagination

APIs return large collections in pages. A generator hides the paging from the caller, who simply loops:

from collections.abc import Iterator


def list_issues(client: httpx.Client, repo: str) -> Iterator[dict]:
    url: str | None = f"/repos/{repo}/issues"
    params: dict | None = {"state": "open", "per_page": 100}
    while url:
        response = request_with_retries(client, "GET", url, params=params)
        yield from response.json()
        url = response.links.get("next", {}).get("url")   # GitHub uses Link headers
        params = None    # the next URL already contains the query string


for issue in list_issues(client, "python/cpython"):
    print(issue["number"], issue["title"])
    if issue["number"] < 100_000:
        break            # pages after this are never fetched

APIs paginate in different ways (a next URL in a header or the body, a cursor token, or page= numbers), but the generator shape is always the same, and stopping early saves requests.

#Validate what comes back with Pydantic

response.json() gives you dicts and lists of whatever the server sent. If the provider renames a field or sends null where you expect a number, your code fails somewhere far from the cause, with a confusing KeyError or TypeError. Validate at the boundary instead, with a Pydantic model:

from datetime import date

from pydantic import BaseModel, ValidationError


class DailyForecast(BaseModel):
    time: list[date]
    temperature_2m_max: list[float]


class ForecastResponse(BaseModel):
    latitude: float
    longitude: float
    daily: DailyForecast


try:
    forecast = ForecastResponse.model_validate(response.json())
except ValidationError as e:
    raise RuntimeError(f"unexpected response from Open-Meteo: {e}") from e

for day, temp in zip(forecast.daily.time, forecast.daily.temperature_2m_max, strict=True):
    print(day.strftime("%a"), f"{temp:.0f}°C")

Pydantic checks the structure, converts values (the "2026-11-01" strings become date objects), ignores extra fields by default, and raises one ValidationError listing every problem. From then on your code works with typed attributes your editor and type checker understand. You'll use the same models to validate incoming requests in the FastAPI lesson.

#Testing without the network

Tests that call real APIs are slow, flaky, need credentials, and can't easily produce a 503 on demand. httpx lets you swap the network layer for a function, with MockTransport:

import httpx
import pytest


def fake_api(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/v1/forecast":
        return httpx.Response(200, json={
            "latitude": 52.5, "longitude": 13.4,
            "daily": {"time": ["2026-11-01"], "temperature_2m_max": [12.5]},
        })
    return httpx.Response(404)


def test_parses_forecast():
    client = httpx.Client(transport=httpx.MockTransport(fake_api), base_url="https://api.open-meteo.com")
    forecast = get_forecast(client, 52.5, 13.4)
    assert forecast.daily.temperature_2m_max == [12.5]

Because your code receives the client as a parameter, the test hands it one backed by a function. The same trick lets you simulate timeouts (raise httpx.ReadTimeout from the handler), 429s and malformed JSON, which are exactly the cases you most need to test and can't easily trigger against a real service.

#Practice

Build a small weather module around the Open-Meteo forecast API (no key needed):

1. get_forecast(client, latitude, longitude) -> ForecastResponse requests daily maximum and minimum temperatures, validates the response with Pydantic, and uses request_with_retries.
2. warmest_day(forecast) -> tuple[date, float] returns the day with the highest maximum.
3. Write tests with MockTransport that check: a normal response is parsed; a 503 followed by a 200 succeeds after one retry; a 404 raises immediately without retrying; a response missing daily raises a clear error. Make retries instant in tests by patching time.sleep.

# src/weather/forecast.py
import random
import time
from datetime import date

import httpx
from pydantic import BaseModel, ValidationError

RETRY_STATUSES = {429, 500, 502, 503, 504}


class Daily(BaseModel):
    time: list[date]
    temperature_2m_max: list[float]
    temperature_2m_min: list[float]


class ForecastResponse(BaseModel):
    latitude: float
    longitude: float
    daily: Daily


class WeatherError(Exception):
    pass


def request_with_retries(client: httpx.Client, method: str, url: str, *, attempts: int = 4, **kwargs) -> httpx.Response:
    for attempt in range(1, attempts + 1):
        try:
            response = client.request(method, url, **kwargs)
            if response.status_code not in RETRY_STATUSES:
                response.raise_for_status()
                return response
            failure: Exception = httpx.HTTPStatusError(
                f"{response.status_code}", request=response.request, response=response
            )
        except httpx.TransportError as e:
            failure = e
        if attempt == attempts:
            raise failure
        time.sleep(random.uniform(0, 0.5 * 2**attempt))
    raise AssertionError("unreachable")


def get_forecast(client: httpx.Client, latitude: float, longitude: float) -> ForecastResponse:
    response = request_with_retries(
        client, "GET", "/v1/forecast",
        params={"latitude": latitude, "longitude": longitude,
                "daily": "temperature_2m_max,temperature_2m_min", "timezone": "auto"},
    )
    try:
        return ForecastResponse.model_validate(response.json())
    except ValidationError as e:
        raise WeatherError(f"unexpected forecast format: {e}") from e


def warmest_day(forecast: ForecastResponse) -> tuple[date, float]:
    pairs = zip(forecast.daily.time, forecast.daily.temperature_2m_max, strict=True)
    return max(pairs, key=lambda pair: pair[1])

# tests/test_forecast.py
from datetime import date

import httpx
import pytest

from weather import forecast as fc

GOOD = {
    "latitude": 52.5, "longitude": 13.4,
    "daily": {"time": ["2026-11-01", "2026-11-02"], "temperature_2m_max": [11.0, 14.5],
              "temperature_2m_min": [4.0, 6.0]},
}


def client_for(*responses: httpx.Response) -> tuple[httpx.Client, list[httpx.Request]]:
    queue, seen = list(responses), []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        return queue.pop(0)

    client = httpx.Client(transport=httpx.MockTransport(handler), base_url="https://api.open-meteo.com")
    return client, seen


@pytest.fixture(autouse=True)
def no_sleep(monkeypatch):
    monkeypatch.setattr(fc.time, "sleep", lambda _: None)


def test_parses_and_finds_warmest_day():
    client, _ = client_for(httpx.Response(200, json=GOOD))
    assert fc.warmest_day(fc.get_forecast(client, 52.5, 13.4)) == (date(2026, 11, 2), 14.5)


def test_retries_after_503():
    client, seen = client_for(httpx.Response(503), httpx.Response(200, json=GOOD))
    fc.get_forecast(client, 52.5, 13.4)
    assert len(seen) == 2


def test_404_is_not_retried():
    client, seen = client_for(httpx.Response(404))
    with pytest.raises(httpx.HTTPStatusError):
        fc.get_forecast(client, 52.5, 13.4)
    assert len(seen) == 1


def test_bad_format_raises_weather_error():
    client, _ = client_for(httpx.Response(200, json={"latitude": 1, "longitude": 2}))
    with pytest.raises(fc.WeatherError, match="unexpected forecast format"):
        fc.get_forecast(client, 52.5, 13.4)

#Key takeaways

• Use an httpx.Client per service with a base_url, default headers and explicit timeouts; build query strings with params.
• Call raise_for_status(), and treat transport errors, 4xx and 5xx/429 differently: only the last group and transport errors are worth retrying.
• Retry with capped exponential backoff and jitter, honour Retry-After, and never blindly retry non-idempotent requests.
• Wrap pagination in a generator, read secrets from the environment, and validate responses at the boundary with Pydantic.
• Pass the client in, and test every failure mode offline with httpx.MockTransport.