Guía completa de asyncio para desarrolladores Python

Asyncio no es solo await y async. Es un modelo de concurrencia cooperativa que, bien usado, mejora rendimiento y escalabilidad en I/O intensivo. Aquí tienes las piezas clave, patrones prácticos, ejemplos completos y por qué funcionan.

1. Conceptos esenciales

• Coroutine: función definida con async def. Se ejecuta cuando el event loop la programa.
• Task: corutina envuelta y schedulada para ejecución concurrente con asyncio.create_task().
• Event loop: motor que ejecuta corutinas y gestiona I/O.
• Concurrencia vs Paralelismo: asyncio ofrece concurrencia cooperativa (útil para I/O). Para CPU-bound se usan procesos o hilos.

2. Ejemplos básicos

Corutina simple y scheduling:

import asyncio

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    await say_after(1, 'hello')
    await say_after(2, 'world')

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

Usando create_task para concurrencia:

import asyncio

async def say(when, what):
    await asyncio.sleep(when)
    return what

async def main():
    t1 = asyncio.create_task(say(2, 'A'))
    t2 = asyncio.create_task(say(1, 'B'))

    print('Tasks creadas')
    res = await asyncio.gather(t1, t2)
    print(res)

asyncio.run(main())

Por qué: create_task permite que ambas corutinas se solapen; gather espera todas y devuelve resultados.

3. Manejo de timeouts y cancelaciones

Timeout seguro con asyncio.wait_for:

import asyncio

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

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=2)
    except asyncio.TimeoutError:
        print('timeout!')

asyncio.run(main())

Cancelación de tasks (propagación correcta):

import asyncio

async def worker():
    try:
        while True:
            print('working...')
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print('limpieza antes de salir')
        raise

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(3)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print('worker cancelado')

asyncio.run(main())

Por qué: necesitas capturar CancelledError si quieres liberar recursos (sockets, archivos, transacciones) antes de terminar.

4. Patrón: Concurrencia limitada (bounded parallelism)

Si lanzas cientos de requests concurridos sin control puedes agotar sockets o APIs. Usa asyncio.Semaphore:

import asyncio
import aiohttp

sem = asyncio.Semaphore(10)

async def fetch(session, url):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

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

Por qué: el semáforo limita el número de conexiones activas a 10, evitando saturar recursos.

5. Patron: Producer-Consumer con asyncio.Queue

import asyncio

async def producer(q, n):
    for i in range(n):
        await q.put(i)
        print('producido', i)
    await q.put(None)  # sentinel

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        print('consumido', item)
        q.task_done()

async def main():
    q = asyncio.Queue()
    p = asyncio.create_task(producer(q, 5))
    c = asyncio.create_task(consumer(q))
    await asyncio.gather(p)
    await q.join()
    await c

asyncio.run(main())

Por qué: Queue maneja sincronización entre productores y consumidores sin bloquear el event loop.

6. I/O bloqueante: usar run_in_executor

Si necesitas usar una librería síncrona (por ejemplo procesamiento de archivos) y no quieres bloquear el loop:

import asyncio
import time

def blocking_io():
    time.sleep(2)
    return 'done'

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_io)  # None -> ThreadPoolExecutor por defecto
    print(result)

asyncio.run(main())

Por qué: delegas bloqueo a un hilo, manteniendo el loop responsivo. Para CPU-bound, usa ProcessPoolExecutor o librerías de multiprocesamiento.

7. Ejemplo práctico: cliente HTTP concurrente con aiohttp

Proyecto pequeño que descarga múltiples páginas y muestra tiempos.

import asyncio
import aiohttp
import time

URLS = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/3',
]

async def fetch(session, url):
    start = time.perf_counter()
    async with session.get(url) as r:
        await r.text()
    elapsed = time.perf_counter() - start
    print(f'{url} -> {elapsed:.2f}s')

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in URLS]
        await asyncio.gather(*tasks)

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

8. Estructura de proyecto recomendada

my_async_project/
├─ src/
│  ├─ __init__.py
│  ├─ app.py           # entrypoint con asyncio.run(main())
│  ├─ http_client.py   # lógica de aiohttp y patterns
│  └─ workers.py       # producers/consumers, tareas background
├─ tests/
│  └─ test_http.py
├─ pyproject.toml
└─ README.md

Por qué: separar entrypoint, lógica de I/O y workers facilita testing y evita ejecutar loop en import time.

9. Errores comunes

• Usar asyncio.run dentro de un loop ya activo (ej. notebooks). Usa nest_asyncio solo como recurso temporal o usa herramientas específicas del entorno.
• No cancelar tareas al apagar la app: provoca fugas y recursos abiertos.
• Usar await en vez de create_task cuando quieres concurrencia: bloqueas hasta que termine.
• Ignorar excepciones de tasks: usar return_exceptions=True en gather o manejar excepciones por task.

10. Integración con frameworks

FastAPI y aiohttp web server ya son asíncronos; no ejecutes operaciones bloqueantes en handlers. Si usas DB síncrona considera un pool de hilos o migrar a asyncpg/Databases.

11. Debugging y herramientas

• Habilita PYTHONASYNCIODEBUG=1 para detectar tareas no esperadas.
• Usa asyncio.all_tasks() para listar tasks y detectar fugas.
• En perfiles de rendimiento, combina sampling profiler con trazas del event loop.

12. Cuándo NO usar asyncio

No es la mejor elección para CPU-bound puro; si tu aplicación es altamente CPU intensiva, usa multiprocesos o librerías específicas (NumPy, C-extensions). Tampoco convierte código síncrono en asíncrono por arte de magia: requiere diseño y pruebas.

Consejo avanzado: si necesitas combinar I/O intensivo con CPU-bound en el mismo servicio, diseña un pipeline híbrido: workers asíncronos que encolan tareas a un pool de procesos (ProcessPoolExecutor) o a un servicio de workers (Celery/RQ/sidekiq). Y siempre instrumenta métricas de latencia y conteo de tasks pendientes para detectar backpressure a tiempo.

Advertencia: no ignores cancelaciones y timeouts en producción — una tarea huérfana puede mantener conexiones y sockets abiertos durante mucho tiempo.

Siguiente paso sugerido: implementa un pequeño microservicio asíncrono que consuma una cola de mensajes, aplique I/O concurrente y delegue operación CPU-bound a procesos; mide latencias y compara con una versión totalmente síncrona.