Build defensive synchronous and asynchronous HTTP clients with explicit timeouts and useful errors.

Requests and HTTPX

Network calls are failure boundaries. DNS can fail, connections can stall, peers can return malformed JSON, and successful HTTP responses can still violate your expected schema. A robust client makes those outcomes explicit.

Choose a client

Requirement	Requests	HTTPX
Straightforward synchronous calls	Excellent	Excellent
Async API	No	Yes
HTTP/2 support	No	Optional
Requests-like interface	Native	Similar
ASGI in-process test transport	No	Yes

Do not choose async merely because it is available. Async helps when one process waits on many independent I/O operations. It does not make CPU-heavy work faster.

Install

uv add requests
# Or:
uv add httpx

A defensive Requests client

from dataclasses import dataclass
from typing import Any

import requests


class CatalogError(RuntimeError):
    """Base error exposed by the catalog adapter."""


class CatalogUnavailable(CatalogError):
    """The catalog could not be reached or returned a retryable failure."""


class CatalogResponseError(CatalogError):
    """The catalog response did not satisfy the expected contract."""


@dataclass(frozen=True, slots=True)
class Product:
    id: int
    name: str


def _parse_product(value: Any) -> Product:
    if not isinstance(value, dict):
        raise CatalogResponseError("product response must be an object")
    product_id = value.get("id")
    name = value.get("name")
    if not isinstance(product_id, int) or not isinstance(name, str):
        raise CatalogResponseError("product requires integer id and string name")
    return Product(id=product_id, name=name)


class CatalogClient:
    def __init__(self, base_url: str, session: requests.Session | None = None) -> None:
        self._base_url = base_url.rstrip("/")
        self._session = session or requests.Session()

    def get_product(self, product_id: int) -> Product:
        try:
            response = self._session.get(
                f"{self._base_url}/products/{product_id}",
                headers={"Accept": "application/json"},
                timeout=(3.05, 10.0),
            )
            response.raise_for_status()
        except requests.Timeout as exc:
            raise CatalogUnavailable("catalog request timed out") from exc
        except requests.ConnectionError as exc:
            raise CatalogUnavailable("catalog connection failed") from exc
        except requests.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else "unknown"
            raise CatalogResponseError(f"catalog returned HTTP {status}") from exc

        try:
            payload = response.json()
        except requests.exceptions.JSONDecodeError as exc:
            raise CatalogResponseError("catalog returned invalid JSON") from exc
        return _parse_product(payload)

The timeout tuple is (connect, read). It is not a total deadline for the complete operation. Requests has no default timeout, so every call should set one.

Reuse a Session for connection pooling and shared headers. Decide who owns it before adding a close() method or context-manager support.

A synchronous HTTPX client

import httpx


class HttpxCatalogClient:
    def __init__(self, base_url: str) -> None:
        self._client = httpx.Client(
            base_url=base_url,
            headers={"Accept": "application/json"},
            timeout=httpx.Timeout(10.0, connect=3.0),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )

    def __enter__(self) -> "HttpxCatalogClient":
        return self

    def __exit__(self, *args: object) -> None:
        self._client.close()

    def get_product(self, product_id: int) -> Product:
        try:
            response = self._client.get(f"/products/{product_id}")
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            raise CatalogUnavailable("catalog request timed out") from exc
        except httpx.NetworkError as exc:
            raise CatalogUnavailable("catalog network failure") from exc
        except httpx.HTTPStatusError as exc:
            raise CatalogResponseError(
                f"catalog returned HTTP {exc.response.status_code}"
            ) from exc

        try:
            return _parse_product(response.json())
        except ValueError as exc:
            raise CatalogResponseError("catalog returned invalid JSON") from exc

Use a context manager:

with HttpxCatalogClient("https://catalog.example") as catalog:
    product = catalog.get_product(42)

Async HTTPX

class AsyncCatalogClient:
    def __init__(self, base_url: str) -> None:
        self._client = httpx.AsyncClient(
            base_url=base_url,
            timeout=httpx.Timeout(10.0, connect=3.0),
        )

    async def __aenter__(self) -> "AsyncCatalogClient":
        return self

    async def __aexit__(self, *args: object) -> None:
        await self._client.aclose()

    async def get_product(self, product_id: int) -> Product:
        try:
            response = await self._client.get(f"/products/{product_id}")
            response.raise_for_status()
        except httpx.TimeoutException as exc:
            raise CatalogUnavailable("catalog request timed out") from exc
        except httpx.NetworkError as exc:
            raise CatalogUnavailable("catalog network failure") from exc
        except httpx.HTTPStatusError as exc:
            raise CatalogResponseError(
                f"catalog returned HTTP {exc.response.status_code}"
            ) from exc

        try:
            return _parse_product(response.json())
        except ValueError as exc:
            raise CatalogResponseError("catalog returned invalid JSON") from exc

Create one long-lived client per service lifecycle. Creating a client for every request defeats connection pooling and can exhaust sockets.

Retry only safe operations

Retries can amplify an outage and duplicate writes. Before retrying, answer:

Is the operation idempotent?
Is the failure transient?
Does the server support an idempotency key?
What is the maximum attempt count and total time budget?
Is there exponential backoff with jitter?
Should Retry-After be honored?

GET requests may be retryable for connection failures, timeouts, 429, and selected 5xx responses. Do not retry authentication failures or malformed requests. Do not blindly retry POST requests that charge money or create records.

Keep error messages safe

Log the method, host, path template, status, elapsed time, and correlation ID. Do not log:

authorization headers;
full URLs containing query secrets;
personal response bodies;
session cookies;
unbounded payloads.

Expose domain errors to the rest of your application so callers do not depend on one HTTP library's exception hierarchy.

Test without real internet

HTTPX supports custom transports:

def handler(request: httpx.Request) -> httpx.Response:
    assert request.url.path == "/products/42"
    return httpx.Response(200, json={"id": 42, "name": "Keyboard"})


def test_catalog_parses_product() -> None:
    transport = httpx.MockTransport(handler)
    with httpx.Client(
        base_url="https://catalog.test",
        transport=transport,
    ) as client:
        response = client.get("/products/42")

    assert _parse_product(response.json()) == Product(id=42, name="Keyboard")

Inject the client or transport into your adapter in production code so the test exercises the public method rather than duplicating its internals.

Common pitfalls

Omitting timeouts or setting only a large undifferentiated timeout.
Calling blocking Requests code directly inside an async endpoint.
Checking status_code == 200 when other 2xx statuses are valid.
Calling .json() before handling error status and content type expectations.
Returning raw dictionaries without validating their shape.
Catching Exception, which turns programming errors into misleading network errors.
Retrying all failures with no attempt or time budget.

Practice

Implement one adapter with explicit connect and read timeouts.
Map transport, timeout, HTTP status, and decoding failures to domain-specific errors.
Test success, 404, 503, invalid JSON, and timeout behavior.
Add a safe retry policy for GET only and test its attempt limit.
Convert a synchronous fan-out operation to AsyncClient and measure whether concurrency helps.

Requests and HTTPX

Requests and HTTPX

Choose a client

Install

A defensive Requests client

A synchronous HTTPX client

Async HTTPX

Retry only safe operations

Keep error messages safe

Test without real internet

Common pitfalls

Practice

Finish this lesson

Requests and HTTPX

On this page