httpx en Python: cliente HTTP moderno con soporte async, HTTP/2 y middleware

httpx es el cliente HTTP de nueva generación para Python. Donde requests ofrece solo una API síncrona, httpx añade una API asíncrona idéntica, soporte nativo de HTTP/2, timeouts configurables por fase, manejo de redireccionamientos, autenticación personalizada y un sistema de transporte intercambiable que facilita enormemente los tests. Si escribes código moderno con asyncio o FastAPI, httpx es la elección natural.

Instalación

# pip install httpx
# pip install httpx[http2]   # para soporte HTTP/2

httpx.Client(): sesión síncrona reutilizable

import httpx

# Usar Client como context manager garantiza que las conexiones se cierren
with httpx.Client(base_url="https://jsonplaceholder.typicode.com") as cliente:
    respuesta = cliente.get("/posts/1")
    respuesta.raise_for_status()   # HTTPStatusError si 4xx/5xx
    datos = respuesta.json()
    print(datos['title'])

    # Parámetros de query
    respuesta2 = cliente.get("/posts", params={"userId": 1, "_limit": 3})
    posts = respuesta2.json()
    print(f"Posts obtenidos: {len(posts)}")

    # POST con JSON
    nuevo = cliente.post(
        "/posts",
        json={"title": "Mi post", "body": "Contenido", "userId": 1}
    )
    print(nuevo.status_code)   # 201
    print(nuevo.json())

AsyncClient: peticiones asíncronas

import asyncio
import httpx

async def obtener_posts(ids: list[int]) -> list[dict]:
    async with httpx.AsyncClient(base_url="https://jsonplaceholder.typicode.com") as cliente:
        tareas = [cliente.get(f"/posts/{i}") for i in ids]
        respuestas = await asyncio.gather(*tareas)
        return [r.json() for r in respuestas]


async def main():
    posts = await obtener_posts([1, 2, 3, 4, 5])
    for post in posts:
        print(f"  [{post['id']}] {post['title']}")

asyncio.run(main())

Timeouts por fase

httpx permite configurar tiempos de espera independientes para cada fase de la conexión:

import httpx

timeout = httpx.Timeout(
    connect=5.0,   # tiempo para establecer la conexión TCP
    read=30.0,     # tiempo entre bytes de la respuesta
    write=10.0,    # tiempo para enviar la petición
    pool=5.0       # tiempo de espera por una conexión del pool
)

with httpx.Client(timeout=timeout) as cliente:
    try:
        resp = cliente.get("https://httpbin.org/delay/2")
        print(resp.status_code)
    except httpx.TimeoutException as e:
        print(f"Timeout: {e}")
    except httpx.ConnectError as e:
        print(f"Error de conexión: {e}")

HTTP/2

import httpx

with httpx.Client(http2=True) as cliente:
    respuesta = cliente.get("https://www.cloudflare.com")
    print(respuesta.http_version)   # HTTP/2
    print(respuesta.status_code)

Autenticación

import httpx

# Basic auth
with httpx.Client(auth=("usuario", "contraseña")) as cliente:
    resp = cliente.get("https://httpbin.org/basic-auth/usuario/contraseña")
    print(resp.json())   # {"authenticated": true, "user": "usuario"}

# Bearer token personalizado
class BearerAuth(httpx.Auth):
    def __init__(self, token: str):
        self.token = token

    def auth_flow(self, request):
        request.headers['Authorization'] = f"Bearer {self.token}"
        yield request


with httpx.Client(auth=BearerAuth("mi_token_secreto")) as cliente:
    resp = cliente.get("https://httpbin.org/bearer")
    print(resp.json())

Cabeceras y cookies persistentes

import httpx

cabeceras = {
    "User-Agent": "MiApp/1.0",
    "Accept-Language": "es-ES,es;q=0.9",
    "X-API-Key": "clave-secreta"
}

with httpx.Client(headers=cabeceras, follow_redirects=True) as cliente:
    # Las cabeceras se envían en todas las peticiones
    resp1 = cliente.get("https://httpbin.org/headers")
    resp2 = cliente.get("https://httpbin.org/get", params={"q": "python"})
    print(resp1.json()['headers']['X-Api-Key'])   # clave-secreta

MockTransport: tests sin servidor real

El sistema de transporte intercambiable de httpx hace que los tests sean triviales: sustituyes el transporte HTTP real por uno que devuelve respuestas predefinidas.

import httpx

def manejador(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/usuarios":
        return httpx.Response(200, json=[{"id": 1, "nombre": "Ana"}])
    if request.url.path.startswith("/usuarios/"):
        uid = int(request.url.path.split("/")[-1])
        return httpx.Response(200, json={"id": uid, "nombre": "Carlos"})
    return httpx.Response(404, json={"error": "not found"})


# En los tests, inyectas el transporte mock
transporte = httpx.MockTransport(manejador)
cliente = httpx.Client(transport=transporte, base_url="https://api.ejemplo.com")

resp = cliente.get("/usuarios")
print(resp.json())   # [{'id': 1, 'nombre': 'Ana'}]

resp2 = cliente.get("/usuarios/42")
print(resp2.json())  # {'id': 42, 'nombre': 'Carlos'}

Reintentos con tenacity

# pip install tenacity
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(httpx.TransportError)
)
def obtener_con_reintento(url: str) -> dict:
    with httpx.Client(timeout=5.0) as cliente:
        resp = cliente.get(url)
        resp.raise_for_status()
        return resp.json()

httpx es una de las pocas bibliotecas Python donde la API síncrona y la asíncrona son prácticamente idénticas, lo que facilita migrar código o mantener ambas versiones sin duplicar lógica.