Peticiones HTTP concurrentes con asyncio.gather y aiohttp

Pedir 100 endpoints uno tras otro te cuesta 100 viajes de latencia. asyncio.gather los lanza a la vez, así que el tiempo total se reduce aproximadamente al de la petición más lenta. Lo que sigue cubre el fetching concurrente con aiohttp: manejo de errores, un semáforo para el rate limiting, timeouts y pool de conexiones.

Cuándo Usar Esto

• Traerte datos de varias APIs en un mismo viaje
• Web scraping con descargas concurrentes de páginas
• Procesamiento por lotes de tareas basadas en HTTP (p. ej., llamar a 100 endpoints de LLM)
• Cualquier carga de trabajo I/O-bound donde las peticiones sean independientes

Requisitos Previos

• Python 3.10+
• El paquete aiohttp (pip install aiohttp)

Solución

Una versión consolidada y ejecutable está en el repositorio complementario.

1. Instalar dependencias

pip install aiohttp

2. Fetch concurrente básico

import asyncio
import aiohttp
import time

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    """Fetch a single URL and return status + content.

    Args:
        session: aiohttp client session.
        url: URL to fetch.

    Returns:
        Dict with url, status, and text.
    """
    async with session.get(url) as response:
        return {
            "url": url,
            "status": response.status,
            "text": await response.text(),
        }

async def fetch_all(urls: list[str]) -> list[dict]:
    """Fetch all URLs concurrently.

    Args:
        urls: List of URLs to fetch.

    Returns:
        List of result dicts in the same order as input URLs.
    """
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Uso
urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

start = time.time()
results = asyncio.run(fetch_all(urls))
elapsed = time.time() - start

print(f"Fetched {len(urls)} URLs in {elapsed:.2f}s (concurrent)")
# ~1.2s en lugar de ~5s en secuencial

3. Manejo de errores con return_exceptions

async def fetch_all_safe(urls: list[str]) -> list[dict | Exception]:
    """Fetch all URLs, capturing exceptions instead of failing.

    Args:
        urls: List of URLs.

    Returns:
        List of results or Exception objects.
    """
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        # return_exceptions=True evita que un fallo cancele a las demás
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# Procesar resultados, manejando excepciones
results = asyncio.run(fetch_all_safe(urls))
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"URL {i} failed: {result}")
    else:
        print(f"URL {i}: status {result['status']}")

4. Limitar la concurrencia con un semáforo

async def fetch_with_limit(
    urls: list[str],
    max_concurrent: int = 10,
) -> list[dict | Exception]:
    """Fetch URLs with a concurrency limit to avoid overwhelming servers.

    Args:
        urls: List of URLs.
        max_concurrent: Maximum simultaneous requests.

    Returns:
        List of results or exceptions.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(session: aiohttp.ClientSession, url: str) -> dict:
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# Limitar a 10 peticiones concurrentes
results = asyncio.run(fetch_with_limit(urls, max_concurrent=10))

5. Timeouts

import aiohttp

async def fetch_with_timeout(
    session: aiohttp.ClientSession,
    url: str,
    timeout_seconds: float = 10.0,
) -> dict:
    """Fetch with a per-request timeout.

    Args:
        session: aiohttp session.
        url: URL to fetch.
        timeout_seconds: Timeout in seconds.

    Returns:
        Result dict or timeout error.
    """
    timeout = aiohttp.ClientTimeout(total=timeout_seconds)
    try:
        async with session.get(url, timeout=timeout) as response:
            return {
                "url": url,
                "status": response.status,
                "text": await response.text(),
            }
    except asyncio.TimeoutError:
        return {"url": url, "status": 0, "error": "timeout"}
    except aiohttp.ClientError as e:
        return {"url": url, "status": 0, "error": str(e)}

async def fetch_all_with_timeouts(urls: list[str], timeout: float = 10.0) -> list[dict]:
    """Fetch all URLs with timeouts."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_timeout(session, url, timeout) for url in urls]
        return await asyncio.gather(*tasks)

6. Pool de conexiones con configuración de sesión

async def fetch_with_pool(
    urls: list[str],
    max_concurrent: int = 20,
) -> list[dict | Exception]:
    """Fetch with optimized connection pool settings.

    Args:
        urls: List of URLs.
        max_concurrent: Max concurrent requests.

    Returns:
        List of results.
    """
    # Configurar el pool de conexiones
    connector = aiohttp.TCPConnector(
        limit=max_concurrent,       # Límite total de conexiones
        limit_per_host=5,           # Límite por host
        ttl_dns_cache=300,          # TTL de la caché DNS en segundos
        enable_cleanup_closed=True,
    )

    timeout = aiohttp.ClientTimeout(
        total=30,      # Timeout total
        connect=10,    # Timeout de conexión
        sock_read=10,  # Timeout de lectura del socket
    )

    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

7. Seguimiento del progreso

async def fetch_with_progress(urls: list[str]) -> list[dict | Exception]:
    """Fetch URLs with real-time progress tracking."""
    results = [None] * len(urls)
    completed = 0

    async def fetch_and_track(session: aiohttp.ClientSession, index: int, url: str):
        nonlocal completed
        try:
            result = await fetch_url(session, url)
            results[index] = result
        except Exception as e:
            results[index] = e
        finally:
            completed += 1
            print(f"\rProgress: {completed}/{len(urls)}", end="", flush=True)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_and_track(session, i, url) for i, url in enumerate(urls)]
        await asyncio.gather(*tasks)

    print()  # Línea nueva tras el progreso
    return results

Cómo Funciona

1. asyncio.gather(*tasks) dispara todas las corrutinas a la vez. El event loop las intercala, suspendiendo cada tarea en sus await y retomándola cuando la I/O termina.
2. aiohttp.ClientSession es quien guarda el pool de conexiones. Con una única sesión para todas las peticiones, las conexiones TCP se reciclan en lugar de reabrirse en cada llamada.
3. asyncio.Semaphore fija cuántas operaciones pueden estar en vuelo a la vez. Cuando max_concurrent tareas ocupan slots, el resto espera a que se libere uno.
4. return_exceptions=True convierte los fallos en valores dentro de la lista de resultados en lugar de lanzarlos, así un endpoint caído no cancela las otras 99 peticiones en vuelo.
5. ClientTimeout define plazos por petición: total acota todo el intercambio, connect el handshake TCP y sock_read la lectura de la respuesta.

Variantes

Procesamiento por lotes con chunks

async def fetch_in_batches(
    urls: list[str],
    batch_size: int = 50,
) -> list[dict | Exception]:
    """Fetch URLs in batches to control memory and rate."""
    results = []

    for i in range(0, len(urls), batch_size):
        batch = urls[i:i + batch_size]
        batch_results = await fetch_all_safe(batch)
        results.extend(batch_results)
        print(f"Completed batch {i // batch_size + 1}")

    return results

Reintentos con backoff exponencial

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> dict:
    """Fetch with retry and exponential backoff."""
    for attempt in range(max_retries):
        try:
            async with session.get(url) as response:
                if response.status == 429:
                    raise aiohttp.ClientError("Rate limited")
                return {
                    "url": url,
                    "status": response.status,
                    "text": await response.text(),
                }
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                return {"url": url, "error": str(e), "attempts": attempt + 1}
            delay = base_delay * (2 ** attempt)
            print(f"Retry {attempt + 1}/{max_retries} for {url} in {delay}s")
            await asyncio.sleep(delay)

async def fetch_all_with_retry(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_retry(session, url) for url in urls]
        return await asyncio.gather(*tasks)

Peticiones POST con JSON

async def post_json(
    session: aiohttp.ClientSession,
    url: str,
    data: dict,
) -> dict:
    """Send a POST request with JSON body."""
    async with session.post(url, json=data) as response:
        return {
            "url": url,
            "status": response.status,
            "json": await response.json(),
        }

async def post_all(
    url: str,
    payloads: list[dict],
) -> list[dict]:
    """Send multiple POST requests concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [post_json(session, url, payload) for payload in payloads]
        return await asyncio.gather(*tasks)

Usar httpx (alternativa a aiohttp)

import httpx
import asyncio

async def fetch_httpx(urls: list[str]) -> list[dict]:
    """Concurrent fetch using httpx (syncs with requests API)."""
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [
            {"url": str(r.url), "status": r.status_code, "text": r.text}
            for r in responses
        ]

Mejores Prácticas

• Reutiliza ClientSession: crear una nueva por cada petición anula por completo el pooling.
• Define limit_per_host: evita martillear un solo servidor con todo a la vez.
• Acota la concurrencia con un semáforo: respeta el rate limit de la API, no solo tu throughput. La receta de rate limiting con semáforos profundiza en esto.
• Pon timeouts siempre: sin ellos, un servidor colgado puede paralizar toda la aplicación.

Errores Comunes

• Crear una ClientSession nueva por petición: anula el pooling; crea la sesión una vez y compártela.
• Olvidar return_exceptions=True: la primera petición que falla cancela todas las demás en vuelo.
• No limitar la concurrencia: lanzar 10.000 peticiones a la vez sobrecarga tu máquina y el servidor.
• Llamar a asyncio.run() dentro de un event loop activo (Jupyter, FastAPI): lanza RuntimeError; haz await a la corrutina directamente.

Ver También

• Documentación de asyncio.gather: la semántica oficial, incluido cómo funciona return_exceptions.
• Referencia del cliente aiohttp: todo sobre sesiones, conectores y timeouts.
• Guía async de httpx: el cliente alternativo usado en las variantes.
• Python Asyncio en Producción: la guía a fondo sobre supervisión, shutdown y observabilidad.