aiohttp en Python: Cliente HTTP Asíncrono

aiohttp ofrece cliente y servidor HTTP para asyncio. En el cliente, reutiliza ClientSession, que mantiene un pool de conexiones y debe cerrarse correctamente.

import asyncio
import aiohttp


async def buscar_estado(url: str) -> dict:
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()


resultado = asyncio.run(buscar_estado("https://api.example.com/status"))

En una aplicación persistente, crea la sesión al iniciar y ciérrala al terminar. Define timeouts y limita tareas simultáneas mediante semáforo o cola. La concurrencia ilimitada puede saturar el servicio y tu proceso.

Valida estado y contenido antes de confiar en JSON. Reintenta solo fallos transitorios y operaciones seguras. Compara HTTPX en Python y revisa TaskGroup.

La documentación oficial del cliente aiohttp, consultada el 22 de julio de 2026, cubre sesiones, streaming y timeouts. Elige según el contrato, no solo por ser asíncrono.

Instalación y modelo asíncrono

Instala la biblioteca en un entorno virtual con python -m pip install aiohttp. Una corrutina cede el control mientras espera red y permite avanzar a otras tareas. Esto mejora el uso de recursos en cargas con mucho I/O, pero no reduce la latencia del servidor remoto ni acelera trabajo pesado de CPU.

asyncio.run() corresponde al punto de entrada de un script. Dentro de FastAPI, aiohttp, Jupyter u otro entorno con un loop activo, usa await y no inicies otro loop. Las bibliotecas deben exponer funciones asíncronas y dejar la propiedad del loop a la aplicación.

Ciclo de vida de ClientSession

Una sesión mantiene pool de conexiones, cookies, cabeceras y configuración compartida. Crear una por llamada desperdicia conexiones y puede agotar sockets. En un script corto, usa un solo async with; en un servicio persistente, crea la sesión al iniciar y ciérrala al terminar.

import aiohttp


class ClienteCatalogo:
    def __init__(self, base_url: str, token: str):
        self._base_url = base_url
        self._token = token
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self):
        timeout = aiohttp.ClientTimeout(total=15, connect=3, sock_read=10)
        self._session = aiohttp.ClientSession(
            base_url=self._base_url,
            timeout=timeout,
            headers={"Authorization": f"Bearer {self._token}"},
        )
        return self

    async def __aexit__(self, exc_type, exc, tb):
        assert self._session is not None
        await self._session.close()

No registres cabeceras de autorización. Guarda credenciales persistentes en el mecanismo de secretos del despliegue. base_url simplifica rutas relativas, pero las URL absolutas necesitan una política clara.

Los timeouts forman parte del contrato

Sin límites útiles, las tareas pueden esperar demasiado por conexión o cuerpo. ClientTimeout distingue tiempo total, establecimiento, adquisición en el pool y lectura entre bloques. Elige valores a partir del objetivo del servicio y del tamaño esperado, no copies números sin medir.

Captura asyncio.TimeoutError por separado de errores HTTP y datos inválidos. Un timeout deja el resultado incierto. En una escritura, el servidor pudo completar la operación antes de perderse la respuesta. Repetir una compra o creación puede duplicar efectos.

import asyncio
import aiohttp


async def obtener_json(session: aiohttp.ClientSession, ruta: str) -> dict:
    try:
        async with session.get(ruta) as respuesta:
            respuesta.raise_for_status()
            return await respuesta.json(content_type="application/json")
    except asyncio.TimeoutError as error:
        raise RuntimeError("La API superó el plazo") from error
    except aiohttp.ClientResponseError as error:
        raise RuntimeError(f"Respuesta HTTP {error.status}") from error

No incluyas todo el cuerpo de error en mensajes públicos o logs indiscriminados, pues puede contener datos privados.

Parámetros, cabeceras y JSON

Usa params para la query string y json para serializar el cuerpo. Así evitas concatenación manual e informas el tipo correcto. data sirve para formularios y bytes; mezclar opciones puede producir otra petición.

async with session.get("/productos", params={"pagina": 2, "limite": 20}) as respuesta:
    respuesta.raise_for_status()
    productos = await respuesta.json()

async with session.post("/eventos", json={"tipo": "vista", "item_id": 42}) as respuesta:
    respuesta.raise_for_status()

raise_for_status() convierte estados 400 y 500 en ClientResponseError, pero el dominio puede interpretar algunos casos. Un 404 puede representar una ausencia esperada y un 409 puede exigir reconciliación. Traduce estas situaciones en una capa cliente para no propagar detalles HTTP.

Antes de confiar en JSON, verifica estado, limita el cuerpo y valida estructura. Un proxy puede devolver HTML inesperadamente. La validación de schema detecta cambios de contrato, pero no debe filtrar todo el payload en errores.

Limitar la concurrencia

asyncio.gather() programa todas las corrutinas entregadas. Con una lista grande puede iniciar miles de peticiones. Un semáforo limita la región que ocupa recursos.

import asyncio


async def buscar_uno(session, url, limite):
    async with limite:
        async with session.get(url) as respuesta:
            respuesta.raise_for_status()
            return await respuesta.json()


async def buscar_todos(session, urls):
    limite = asyncio.Semaphore(10)
    tareas = [buscar_uno(session, url, limite) for url in urls]
    return await asyncio.gather(*tareas)

El conector también limita conexiones mediante aiohttp.TCPConnector(limit=...). El semáforo expresa concurrencia de operaciones; el conector protege su pool. Para colecciones enormes, usa una cola con workers fijos, porque millones de tareas también consumen memoria. Respeta cuotas y respuestas 429.

Reintentar con responsabilidad

Reintenta solo fallos transitorios, con máximo de intentos, espera exponencial y jitter. GET, HEAD y ciertas operaciones con clave de idempotencia son candidatas. POST no es seguro automáticamente. Estados 400, 401, 403 y fallos de validación suelen requerir corrección.

Si el servicio envía Retry-After, interprétalo dentro de un máximo razonable. La cancelación debe interrumpir espera y petición; no absorbas CancelledError como un fallo normal. Centraliza la política para mantener comportamiento y métricas consistentes.

Streaming con memoria limitada

await response.read() y await response.json() acumulan el cuerpo. Para archivos grandes, itera bloques de response.content e impone un límite. Escribe en temporal y publica solo tras completar y, si existe, verificar un hash.

from pathlib import Path


async def descargar(session, url: str, destino: Path, max_bytes=20_000_000):
    total = 0
    async with session.get(url) as respuesta:
        respuesta.raise_for_status()
        with destino.open("wb") as archivo:
            async for bloque in respuesta.content.iter_chunked(64 * 1024):
                total += len(bloque)
                if total > max_bytes:
                    raise ValueError("La descarga superó el límite")
                archivo.write(bloque)

El código de producción debe eliminar el temporal al fallar. Nunca conviertas el nombre remoto en una ruta local. Genera el destino y evita sobrescrituras.

URL externas y SSRF

Un cliente que consulta una URL proporcionada por usuarios puede permitir SSRF hacia servicios internos, metadatos de nube o localhost. Prefiere una lista de esquemas y hosts permitidos. Desactiva redirecciones o vuelve a validar cada destino.

Comprobar solo el texto del hostname puede ser insuficiente ante DNS y formas alternativas de dirección. Aplicaciones de alto riesgo requieren controles de red, resolución o proxy además de validación en código. Nunca reenvíes autorización de una API confiable a un host arbitrario.

Pruebas, observabilidad y cierre

Prueba éxito, estados de error, JSON inválido, conexión rechazada, timeout, cuerpo excesivo, cancelación y 429. Un servidor HTTP local o una fixture específica reproduce streaming y retrasos mejor que un mock superficial.

Registra método, host aprobado, ruta normalizada, estado, duración, intentos y bytes, sin query strings sensibles ni tokens. Métricas de latencia y error permiten ajustar timeout y concurrencia con evidencia.

Antes de desplegar, confirma una sesión por ciclo de vida, cierre garantizado, timeouts explícitos, concurrencia limitada, liberación de respuesta mediante async with, retries solo en casos seguros y validación de destinos externos. El beneficio de aiohttp nace de coordinar esperas de red con disciplina, no de lanzar el máximo número de tareas.

Define una versión explícita del contrato de API y prueba las cabeceras de negociación. Una integración resistente falla con claridad ante un tipo de medio o schema incompatible, en lugar de aceptar cualquier diccionario JSON. Mantén modelos de parsing separados del transporte para revisar migraciones de contrato sin mezclar responsabilidades.

La conexión también depende de caché DNS, verificación TLS, proxies y configuración del conector. Mantén activa la verificación de certificados. Si una autoridad privada es necesaria, crea un contexto SSL deliberado en vez de desactivar controles. URL y credenciales de proxy son secretos y no deben aparecer en excepciones.

Durante el cierre del servicio, deja de aceptar trabajo, permite que operaciones en curso terminen dentro de un límite y luego cierra la sesión. Ese orden evita avisos de Unclosed client session y trabajos parciales. El apagado elegante también necesita un plazo máximo, para que un proveedor averiado no bloquee indefinidamente el despliegue.

Documenta si las respuestas se cachean y cómo se invalidan. No caches indiscriminadamente datos personales ni respuestas autenticadas. Respeta encabezados y requisitos del dominio, y añade una clave de caché que incluya los parámetros relevantes para impedir que un usuario reciba el resultado de otro.