aiohttp: Peticiones HTTP Concurrentes con asyncio

En el artículo anterior cerramos con una regla clave: nunca uses APIs bloqueantes como requests dentro de una coroutine, porque bloquean el event loop. La alternativa asíncrona por excelencia para clientes HTTP es aiohttp. En este artículo la pondremos en práctica y, más importante aún, veremos cómo coordinar cientos de peticiones concurrentes con las tres APIs que asyncio ofrece para esto: gather, as_completed y wait.

El Patrón async with: Context Managers Asíncronos

Antes de tocar aiohttp debemos entender el patrón sobre el que se construye. Un context manager síncrono se define con __enter__ y __exit__ y nos permite adquirir y liberar recursos de forma segura con with. Su equivalente asíncrono usa los métodos __aenter__ y __aexit__, y se consume con async with.

La diferencia es que estos métodos son coroutines: pueden usar await internamente para adquirir o liberar el recurso sin bloquear el event loop.

class ConnectedSocket:
    def __init__(self, server_socket):
        self._connection = None
        self._server_socket = server_socket

    async def __aenter__(self):                 # 1
        loop = asyncio.get_event_loop()
        connection, address = await loop.sock_accept(self._server_socket)
        self._connection = connection
        return self._connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):  # 2
        self._connection.close()

1. __aenter__ se ejecuta al entrar al bloque async with y retorna el recurso (la conexión).
2. __aexit__ se ejecuta al salir del bloque —incluso si hay una excepción— y libera el recurso.

Esto es relevante porque, como veremos enseguida, tanto la sesión de aiohttp como cada petición que hacemos son context managers asíncronos.

sequenceDiagram
    participant C as Código (async with)
    participant CM as Context Manager
    C->>CM: await __aenter__()
    Note over CM: adquiere el recurso (sin bloquear)
    CM-->>C: retorna el recurso
    Note over C: ejecuta el bloque async with
    C->>CM: await __aexit__()
    Note over CM: libera el recurso

aiohttp: Sesiones y Connection Pooling

aiohttp, y los clientes HTTP en general, se basan en el concepto de sesión. Una ClientSession mantiene abiertas varias conexiones que luego se reutilizan entre peticiones. Esto se conoce como connection pooling y es fundamental para el rendimiento: crear una conexión TCP es costoso, así que reutilizar un pool reduce drásticamente esa sobrecarga.

import asyncio
import aiohttp
from aiohttp import ClientSession


async def fetch_status(session: ClientSession, url: str) -> int:
    async with session.get(url) as result:   # 1
        return result.status


async def main():
    async with aiohttp.ClientSession() as session:  # 2
        url = 'https://example.com'
        status = await fetch_status(session, url)
        print(f'Status para {url}: {status}')


asyncio.run(main())

1. session.get(url) es un context manager asíncrono: dentro del bloque tenemos la respuesta; al salir, la conexión vuelve al pool.
2. La sesión también se gestiona con async with, garantizando que todas las conexiones se cierren correctamente al terminar.

Configurando Timeouts

Por defecto aiohttp aplica un timeout de cinco minutos a cada operación. Para la mayoría de aplicaciones ese valor es demasiado alto. Podemos ajustarlo con ClientTimeout, tanto a nivel de sesión (aplica a todas las peticiones) como a nivel de petición individual.

async def fetch_status(session: ClientSession, url: str) -> int:
    diez_millis = aiohttp.ClientTimeout(total=.01)         # por petición
    async with session.get(url, timeout=diez_millis) as result:
        return result.status


async def main():
    session_timeout = aiohttp.ClientTimeout(total=1, connect=.1)  # por sesión
    async with aiohttp.ClientSession(timeout=session_timeout) as session:
        await fetch_status(session, 'https://example.com')

El parámetro total limita la duración completa de la operación; connect limita solo el tiempo de establecer la conexión. El timeout de la petición tiene prioridad sobre el de la sesión.

La Trampa del await Secuencial

Antes de lanzar muchas peticiones, hay un error sutil que conviene evitar. Supongamos que queremos ejecutar varias coroutines y recolectar resultados con una list comprehension:

async def main() -> None:
    delay_times = [3, 3, 3]
    # await dentro del comprehension → espera cada task antes de crear el siguiente
    [await asyncio.create_task(delay(seconds)) for seconds in delay_times]

async def main() -> None:
    delay_times = [3, 3, 3]
    tasks = [asyncio.create_task(delay(seconds)) for seconds in delay_times]  # 1
    [await task for task in tasks]                                            # 2

La clave está en separar la creación de los tasks (que los pone a correr en el event loop) de la espera de sus resultados. Si haces await mientras los creas, los serializas. Esta separación es precisamente lo que las siguientes APIs resuelven por nosotros.

gather: Recolectar Todos los Resultados

asyncio.gather recibe una secuencia de awaitables, los ejecuta concurrentemente y, cuando hacemos await sobre él, espera a que todos terminen y retorna una lista con sus resultados. Si le pasamos coroutines, las envuelve en tasks automáticamente, así que no necesitamos llamar a create_task manualmente.

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ['https://example.com' for _ in range(1000)]
        requests = [fetch_status(session, url) for url in urls]
        status_codes = await asyncio.gather(*requests)   # 1
        print(status_codes)

1. gather ejecuta las 1000 peticiones de forma concurrente y retorna sus códigos de estado.

Una garantía importante: aunque las peticiones terminen en cualquier orden, los resultados se devuelven en el mismo orden en que pasamos los awaitables. Esto hace que gather sea ideal cuando necesitas correlacionar cada resultado con su entrada.

Manejo de Excepciones en gather

El comportamiento ante errores se controla con el parámetro return_exceptions:

• return_exceptions=False (por defecto): si cualquier awaitable lanza una excepción, esa excepción se propaga al hacer await gather. El resto de coroutines siguen ejecutándose, pero perdemos sus resultados a menos que manejemos la excepción.
• return_exceptions=True: las excepciones se devuelven como parte de la lista de resultados, junto a los valores exitosos. gather nunca lanza, y nosotros decidimos cómo tratar cada error.

results = await asyncio.gather(*requests, return_exceptions=True)

exitos = [r for r in results if not isinstance(r, Exception)]
errores = [r for r in results if isinstance(r, Exception)]

as_completed: Procesar a Medida que Terminan

Cuando queremos reaccionar a cada resultado en cuanto está disponible, asyncio.as_completed es la herramienta. Recibe una lista de awaitables y retorna un iterador de futures. Al iterar y hacer await sobre cada uno, obtenemos los resultados en el orden en que van completándose, no en el que los pasamos.

async def main():
    async with aiohttp.ClientSession() as session:
        fetchers = [
            fetch_status(session, 'https://example.com', delay=1),
            fetch_status(session, 'https://example.com', delay=1),
            fetch_status(session, 'https://example.com', delay=10),
        ]

        for finished_task in asyncio.as_completed(fetchers):
            print(await finished_task)   # se imprime apenas cada uno termina

Las dos primeras peticiones se imprimen alrededor del segundo 1, sin esperar a la de 10 segundos. Esto mejora la responsividad de la aplicación. También nos da mejor control de errores: la excepción de un task se lanza justo cuando hacemos await sobre él, permitiéndonos manejarla en el momento.

as_completed admite un timeout global. Si se excede, el await sobre los futures pendientes lanza asyncio.TimeoutError, que podemos capturar:

for done_task in asyncio.as_completed(fetchers, timeout=2):
    try:
        print(await done_task)
    except asyncio.TimeoutError:
        print('¡Timeout!')

flowchart LR
    subgraph gather["gather → espera TODO"]
        G1[req 1s] --> GW[⏳ bloquea hasta el más lento]
        G2[req 1s] --> GW
        G3[req 10s] --> GW
        GW --> GR[procesa los 3 resultados]
    end
    subgraph asc["as_completed → procesa al terminar"]
        A1[req 1s] --> AR1[procesa ~1s]
        A2[req 1s] --> AR2[procesa ~1s]
        A3[req 10s] --> AR3[procesa ~10s]
    end

La desventaja de as_completed es que no es fácil saber qué tasks ya terminaron y cuáles siguen pendientes: los recibimos uno a uno por el iterador.

wait: Control Fino sobre la Espera

asyncio.wait es la opción más flexible. Recibe una colección de tasks y retorna dos conjuntos: done (los que terminaron, con resultado o excepción) y pending (los que siguen corriendo). A diferencia de las otras APIs, wait no lanza excepciones automáticamente ni cancela tasks por timeout; nos da el control y la responsabilidad.

Su comportamiento se ajusta con el parámetro return_when, que acepta tres valores:

Como wait nos entrega el conjunto pending, podemos cancelar explícitamente los tasks que ya no necesitamos:

async def main():
    async with aiohttp.ClientSession() as session:
        fetchers = [
            asyncio.create_task(fetch_status(session, 'https://example.com')),
            asyncio.create_task(fetch_status(session, 'https://example.com')),
        ]

        done, pending = await asyncio.wait(
            fetchers, return_when=asyncio.FIRST_COMPLETED
        )

        print(f'Terminados: {len(done)} | Pendientes: {len(pending)}')

        for task in done:
            if task.exception() is None:   # 1
                print(task.result())
            else:
                print(f'Error: {task.exception()}')

        for task in pending:
            task.cancel()                  # 2

1. Como los tasks en done ya terminaron, es seguro consultar exception() y result(). Esto evita usar await con try/except.
2. Cancelamos manualmente lo que quedó pendiente —algo que gather no permite hacer con facilidad.

Sobre los timeouts en wait: si pasas el parámetro timeout y se agota, wait simplemente retorna done y pending hasta ese momento, sin lanzar excepción y sin cancelar los tasks pendientes. Tú decides qué hacer con ellos.

¿Cuál Debo Usar?

Las tres APIs ejecutan awaitables concurrentemente; la elección depende de cómo necesites consumir los resultados.

flowchart TD
    Start{¿Cómo necesito
los resultados?} --> A[Todos juntos
y ordenados]
    Start --> B[Cada uno apenas
termina]
    Start --> C[Control fino:
done/pending,
cancelar el resto]

    A --> AG["asyncio.gather"]
    B --> AC["asyncio.as_completed"]
    C --> AW["asyncio.wait"]

    style AG fill:#3b82f6,stroke:#1e40af,color:#fff
    style AC fill:#f59e0b,stroke:#b45309,color:#fff
    style AW fill:#10b981,stroke:#065f46,color:#fff

• gather → la opción más simple cuando quieres todos los resultados, en orden, y un único punto donde manejar el éxito o el error.
• as_completed → cuando quieres procesar cada resultado en cuanto llega y mejorar la responsividad.
• wait → cuando necesitas control fino: saber qué terminó y qué sigue pendiente, manejar errores sin excepciones implícitas y cancelar tasks manualmente.

En conclusión

• aiohttp es el reemplazo asíncrono de requests; usa una ClientSession con connection pooling y reutilízala entre peticiones.
• Tanto la sesión como cada petición son context managers asíncronos (async with), construidos sobre __aenter__/__aexit__.
• Separa la creación de tasks de su espera: hacer await mientras los creas los serializa.
• gather espera todo y devuelve resultados ordenados; controla errores con return_exceptions.
• as_completed entrega resultados a medida que terminan, ideal para responsividad.
• wait ofrece control fino con los conjuntos done/pending y los modos ALL_COMPLETED, FIRST_EXCEPTION y FIRST_COMPLETED.

Referencias

• aiohttp — Documentación Oficial
• Python asyncio.gather — Documentación Oficial
• Python asyncio.as_completed — Documentación Oficial
• Python asyncio.wait — Documentación Oficial
• PEP 492 — Coroutines with async and await syntax
• Matthew Fowler, Python Concurrency with asyncio (Manning, 2022)