# Calling HTTP APIs

> Call web APIs reliably with httpx: clients, timeouts, errors, retries with backoff, pagination, secrets from the environment, Pydantic validation and offline tests.

Python roadmap · Building things · 18 min read · https://levelcode.org/roadmaps/python/http-apis-python

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

```bash
uv add httpx pydantic
```

```python
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()`:

```python
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:

| Failure | httpx exception | Typical response |
|---|---|---|
| No usable response: DNS failure, refused connection, timeout | `httpx.TransportError` (and subclasses such as `httpx.TimeoutException`, `httpx.ConnectError`) | Retry later |
| A response with an error status | `httpx.HTTPStatusError`, with `.response` attached | Depends on the status |

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:

```python
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.

> [!WARNING]
> Never hard-code API keys in source code. They end up in Git history, screenshots and logs. Read them from environment variables (`os.environ["GITHUB_TOKEN"]` fails loudly if it's missing, which is what you want), and keep them out of log messages and exception text.

## 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.

```python
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:

```python
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:

```python
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`:

```python
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`.

**Solution**

```python
# 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])
```

```python
# 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)
```

Recording every request in `seen` turns "did it retry?" into a simple assertion on its length. The autouse fixture patches `time.sleep` in the module under test, so the retry test runs instantly. Wrapping `ValidationError` in a `WeatherError` keeps Pydantic an internal detail: callers handle one exception type that belongs to your module.

## 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`.

## Quiz

1. An API returns 404 for a user ID. Should your code retry?

   A. Yes, with exponential backoff
   B. No: the same request will get the same answer; report or handle the missing resource
   C. Yes, immediately, up to 10 times
   D. Only if the request was a GET

   **Answer: B.** 4xx errors (other than 429) mean the request itself is wrong or refers to something that doesn't exist. Retrying just repeats it.

2. Why is it risky to automatically retry a POST that creates a payment after a timeout?

   A. POST requests can't be retried in httpx
   B. Timeouts always mean the payment failed
   C. Retries are slower than the original request
   D. The first request may have succeeded before the connection dropped, so a retry can charge twice unless the API supports idempotency keys

   **Answer: D.** A timeout tells you that you didn't get an answer, not that nothing happened. Idempotency keys let the server recognise and deduplicate the retry.

3. What does validating a JSON response with a Pydantic model give you?

   A. Faster HTTP requests
   B. Automatic retries
   C. One clear error at the boundary if the shape is wrong, plus typed, converted values afterwards
   D. Encryption of the response

   **Answer: C.** Pydantic checks and converts the data where it enters your program, so problems surface immediately with a precise message instead of as KeyErrors deep inside your code.

4. How can a test check that your code retries after a 503, without touching the network?

   A. Pass the code an httpx.Client backed by httpx.MockTransport that returns 503 and then 200
   B. Temporarily take the real API down
   C. Set a very short timeout
   D. Mock the json() method

   **Answer: A.** MockTransport replaces the network with a function, so the test controls every response, including errors that are hard to trigger for real.