Guía completa de asyncio en Python para desarrolladores

Asyncio es la librería estándar de Python para concurrencia basada en corutinas y un bucle de eventos. Esta guía práctica te lleva desde los conceptos básicos hasta patrones robustos: creación y cancelación de tareas, manejo de errores, timeouts, límites de concurrencia, integración con I/O de red (aiohttp) y cómo trabajar con código CPU-bound.

Estructura del proyecto

asyncio-guide/
├─ examples/
│  ├─ basic_task.py
│  ├─ semaphore_rate_limiter.py
│  ├─ aiohttp_client.py
│  └─ cpu_bound_executor.py
├─ requirements.txt
└─ README.md

requirements.txt (opcional si usas venv con Python 3.11+):

aiohttp==3.8.*

Conceptos clave (rápido)

• Corutina: una función definida con async def.
• Tarea (Task): envoltorio que programa una corutina para ejecutarse en el bucle de eventos (con asyncio.create_task).
• Loop: el ciclo que gestiona ejecución y switching cooperativo.
• Concurrency ≠ Parallelism: asyncio permite ejecutar muchas operaciones I/O-concurrentes; para CPU-bound necesitas hilos/procesos.

1) Ejemplo básico: crear, esperar y cancelar tareas

import asyncio

async def worker(name, delay):
    try:
        print(f"{name}: iniciando, espera {delay}s")
        await asyncio.sleep(delay)
        print(f"{name}: terminado")
    except asyncio.CancelledError:
        print(f"{name}: cancelado")
        raise

async def main():
    t1 = asyncio.create_task(worker('tarea1', 2))
    t2 = asyncio.create_task(worker('tarea2', 5))

    await asyncio.sleep(3)
    t2.cancel()

    await asyncio.gather(t1, t2, return_exceptions=True)

if __name__ == '__main__':
    asyncio.run(main())

Por qué: create_task lanza corutinas en paralelo lógico; cancelar una tarea lanza CancelledError dentro de la corutina, por eso manejar la excepción es importante para limpiar recursos.

2) Timeouts y manejo robusto

import asyncio

async def slow_op():
    await asyncio.sleep(10)

async def main():
    try:
        await asyncio.wait_for(slow_op(), timeout=3)
    except asyncio.TimeoutError:
        print('Operación excedió el timeout')

if __name__ == '__main__':
    asyncio.run(main())

Por qué: wait_for encapsula cancelación y proporciona un límite temporal. Evita usar time.sleep en corutinas: bloquea el loop.

3) Límite de concurrencia: semáforos y pools

import asyncio
import random

semaphore = asyncio.Semaphore(5)  # máximo 5 tareas concurrentes

async def fetch(i):
    async with semaphore:
        delay = random.uniform(0.5, 2.0)
        print(f'fetch {i} start')
        await asyncio.sleep(delay)
        print(f'fetch {i} done')

async def main():
    await asyncio.gather(*(fetch(i) for i in range(50)))

if __name__ == '__main__':
    asyncio.run(main())

Por qué: controlar el grado de concurrencia evita agotar recursos (sockets, CPU, API rate limits).

4) I/O de red: cliente HTTP con aiohttp

import asyncio
import aiohttp

async def fetch(session, url):
    try:
        async with session.get(url, timeout=10) as resp:
            text = await resp.text()
            return resp.status, len(text)
    except asyncio.TimeoutError:
        return 'timeout', 0

async def main(urls):
    conn = aiohttp.TCPConnector(limit=10)  # limite de conexiones
    timeout = aiohttp.ClientTimeout(total=20)
    async with aiohttp.ClientSession(connector=conn, timeout=timeout) as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print(results)

if __name__ == '__main__':
    urls = ['https://example.com'] * 20
    asyncio.run(main(urls))

Por qué: reuse de ClientSession y TCPConnector mejora rendimiento. Evita crear sesión por request. Usa límites en el conector para no saturar la red.

5) Código CPU-bound: run_in_executor

import asyncio
import concurrent.futures
import math

def cpu_heavy(n):
    # ejemplo de tarea que bloquea CPU
    s = 0
    for i in range(1, n):
        s += math.sqrt(i)
    return s

async def main():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as pool:
        # si es CPU-bound, preferir procesos (GIL)
        result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000)
        print('resultado', result)

if __name__ == '__main__':
    asyncio.run(main())

Por qué: run_in_executor mueve trabajo bloqueante fuera del loop; usar ProcessPoolExecutor para bypass del GIL en tareas pesadas.

6) Diferencias útiles: gather vs wait vs as_completed

• asyncio.gather: espera a todas; opcionalmente devuelve excepciones si no usas return_exceptions=True.
• asyncio.wait: controla cuándo retorna (FIRST_COMPLETED / FIRST_EXCEPTION / ALL_COMPLETED) y devuelve sets de done/pending.
• asyncio.as_completed: itera resultados según se completen — útil para streaming de resultados.

7) Manejo de excepciones y cancelaciones en tareas hijas

async def parent():
    async def child():
        try:
            await asyncio.sleep(10)
        finally:
            # limpieza crítica
            print('limpieza child')

    t = asyncio.create_task(child())
    await asyncio.sleep(0.1)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print('parent: child cancelado')

Por qué: siempre pondrás limpieza en finally porque CancelledError puede ser lanzado en cualquier await.

8) Debugging y herramientas

• Habilita debug: asyncio.run(main(), debug=True) o loop.set_debug(True).
• Usa asyncio.all_tasks() para inspeccionar tareas en el loop.
• Herramientas: aiomonitor, python -X dev para detectar rutinas olvidadas.

9) Errores comunes y cómo evitarlos

• Crear tasks en ciclo sin límites -> memory leak / demasiadas conexiones: usa semáforos o pools.
• Olvidar await en una corutina -> no se ejecuta; o crear task sin manejar excepciones -> excepciones silenciadas (usa logging y gather(return_exceptions=True) en pruebas).
• Usar time.sleep en corutinas -> bloquea totalmente el loop.
• No cerrar ClientSession -> sockets en TIME_WAIT (usa async with).
• No manejar CancelledError para limpiar recursos -> fugas y estados inconsistentes.

10) Buenas prácticas resumidas

• Reusar ClientSession y límites de conexiones.
• Controlar concurrencia con Semaphore o pools.
• Separar tareas I/O-bound (asyncio) de CPU-bound (PoolExecutor/ProcessPoolExecutor).
• Manejar cancelaciones y excepciones explícitamente.
• Medir: latency y throughput antes de optimizar.

Archivo ejemplo: semaphore_rate_limiter.py

import asyncio
import aiohttp

SEM_LIMIT = 10
semaphore = asyncio.Semaphore(SEM_LIMIT)

async def fetch(session, url):
    async with semaphore:
        async with session.get(url) as resp:
            return resp.status

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in urls]
        for fut in asyncio.as_completed(tasks):
            print(await fut)

if __name__ == '__main__':
    urls = ['https://httpbin.org/delay/1'] * 50
    asyncio.run(main(urls))

Por qué: usar as_completed permite procesar respuestas conforme llegan, reduciendo latencia aparente y memory spikes.

Pruebas y mediciones

Mide con time.perf_counter alrededor de secciones críticas. Para carga real usa herramientas como wrk/hey o scripts que simulan concurrencia. Mide conexiones abiertas, CPU, latencia p95/p99.

Recursos y siguientes pasos

Lee la documentación oficial de asyncio y ejemplos de aiohttp; experimenta combinando semáforos y as_completed para flujos de trabajo con backpressure.

Consejo avanzado: si necesitas combinar asyncio con bibliotecas sin soporte async (p. ej. una SDK bloqueante), encapsula llamadas en run_in_executor y considera un pool dedicado para no mezclar latencias. Advertencia: mezclar muchos hilos y corutinas sin límites puede volver tu arquitectura más compleja que necesaria; mide antes de escalar horizontalmente.