43. HTTP clients done right: timeouts, retries, pooling, and an SSRF guard

43.1 requests and httpx

requests is the classic synchronous client (Part VIII reads its source). httpx has almost the same API, plus async support and HTTP/2. Both are fine; use httpx when you need async.

import httpx

with httpx.Client(
    base_url="https://api.github.com",
    timeout=httpx.Timeout(10.0, connect=3.0),       # ALWAYS set timeouts
    headers={"User-Agent": "py-labs/0.1"},
    follow_redirects=True,
) as client:
    r = client.get("/repos/psf/requests")
    r.raise_for_status()                           # 4xx/5xx -> exception
    repo = r.json()
    print(repo["full_name"], repo["stargazers_count"])

The rules that matter in production:

1. Always set a timeout. requests has no default timeout: a server that accepts the connection and never answers hangs your worker forever. httpx defaults to 5 seconds. Set both a connect timeout (is the host there?) and a read timeout (is it answering?).
2. Reuse a client/session. Each requests.get() creates a new session, a new connection pool, a new TCP handshake, and a new TLS handshake. A Session/Client keeps connections alive and reuses them. For 100 calls to one host, that is the difference between 100 handshakes and 1 (Net Ch 21, Sec Ch 28).
3. Check the status. raise_for_status() or an explicit check; a 500 page parsed as JSON produces confusing errors later.
4. Never verify=False outside a throwaway local test. It disables certificate checking and makes every request interceptable.

43.2 Retries with backoff, and idempotency keys

import random
import time
import uuid
import httpx

RETRYABLE = {429, 502, 503, 504}

def post_with_retries(client: httpx.Client, url: str, payload: dict, attempts: int = 4) -> httpx.Response:
    key = str(uuid.uuid4())                       # same key on every retry of THIS operation
    for attempt in range(1, attempts + 1):
        try:
            r = client.post(url, json=payload, headers={"Idempotency-Key": key})
            if r.status_code not in RETRYABLE:
                return r
            retry_after = float(r.headers.get("retry-after", 0) or 0)
        except (httpx.ConnectError, httpx.ReadTimeout):
            retry_after = 0
            if attempt == attempts:
                raise
        if attempt == attempts:
            return r
        delay = max(retry_after, min(8.0, 0.25 * 2 ** attempt) * random.uniform(0.5, 1.0))
        time.sleep(delay)                         # exponential backoff + jitter, honoring Retry-After
    raise AssertionError("unreachable")

The idempotency key lets the server recognize a retry of a POST it already processed and return the original result instead of charging the customer twice. The HTTPS guide Ch 21 explains the server side, retry budgets, and circuit breakers. For simple cases, httpx.HTTPTransport(retries=3) retries connection failures, and urllib3.Retry mounted on a requests session (Ch 65) does status-based retries.

43.3 Async: many requests at once

import asyncio
import httpx

async def check(client: httpx.AsyncClient, url: str) -> tuple[str, int | str]:
    try:
        r = await client.get(url)
        return url, r.status_code
    except httpx.HTTPError as e:
        return url, type(e).__name__

async def main(urls: list[str]) -> None:
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with httpx.AsyncClient(timeout=5, limits=limits) as client:
        for url, status in await asyncio.gather(*(check(client, u) for u in urls)):
            print(status, url)

asyncio.run(main(["https://example.com", "https://www.python.org", "https://nonexistent.invalid"]))

43.4 An SSRF guard

If your service fetches URLs that users provide (webhooks, "import from URL", link previews, an agent's fetch tool), an attacker will give it http://169.254.169.254/latest/meta-data/ (the cloud metadata service) or http://localhost:6379/. That is SSRF (Security in Depth Ch 33, HTTPS Ch 22).

# labs/safefetch.py
import ipaddress
import socket
from urllib.parse import urlsplit
import httpx

class BlockedURL(Exception):
    pass

def _check_ip(ip: str) -> None:
    addr = ipaddress.ip_address(ip)
    if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped:
        addr = addr.ipv4_mapped                     # ::ffff:127.0.0.1 is 127.0.0.1
    if not addr.is_global or addr.is_multicast:
        raise BlockedURL(f"non-public address {addr}")

def resolve_public(url: str) -> tuple[str, str, int]:
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https") or not parts.hostname:
        raise BlockedURL("only http(s) URLs with a host")
    port = parts.port or (443 if parts.scheme == "https" else 80)
    if port not in (80, 443):
        raise BlockedURL(f"port {port} not allowed")
    infos = socket.getaddrinfo(parts.hostname, port, type=socket.SOCK_STREAM)
    for *_, sockaddr in infos:                     # EVERY address must be public
        _check_ip(sockaddr[0])
    return parts.hostname, infos[0][4][0], port

def safe_get(url: str, max_bytes: int = 2 * 2**20) -> bytes:
    host, ip, port = resolve_public(url)
    # Connect to the IP we just checked, not to the name: a second DNS lookup could return a
    # different, private address (DNS rebinding). Keep the real name for Host and for TLS (SNI and
    # certificate verification) so HTTPS still validates against the hostname.
    netloc = f"[{ip}]:{port}" if ":" in ip else f"{ip}:{port}"
    pinned = urlsplit(url)._replace(netloc=netloc).geturl()
    with httpx.Client(timeout=5, follow_redirects=False) as c:
        with c.stream("GET", pinned, headers={"Host": host}, extensions={"sni_hostname": host}) as r:
            if r.is_redirect:
                raise BlockedURL("redirects must be re-validated; refusing")
            data = b""
            for chunk in r.iter_bytes():
                data += chunk
                if len(data) > max_bytes:
                    raise BlockedURL("response too large")
            return data

if __name__ == "__main__":
    for u in ["http://169.254.169.254/latest/meta-data/", "http://localhost:6379/",
              "http://[::ffff:127.0.0.1]/", "ftp://example.com/", "https://example.com/"]:
        try:
            print("OK     ", u, len(safe_get(u)), "bytes")
        except (BlockedURL, httpx.HTTPError, OSError) as e:
            print("BLOCKED", u, "-", e)

The checks that matter: allow-list schemes and ports; resolve the name and reject any non-global address (loopback, private, link-local, the metadata range, IPv4-mapped IPv6); connect to the address you checked, so DNS rebinding cannot swap it; do not follow redirects without re-checking; cap the response size and time. Large companies add one more layer: an egress proxy that enforces the same policy at the network edge, so a bug in one service's guard is not enough.