Guía completa de asyncio para desarrolladores Python

Esta guía práctica te llevará desde los conceptos básicos de asyncio hasta patrones de producción: tareas, concurrencia controlada, manejo de errores, integración con código bloqueante y un ejemplo completo: un crawler asíncrono eficiente.

¿Qué es asyncio y cuándo usarlo?

asyncio es la librería estándar de Python para programación asíncrona basada en corrutinas (coroutines) y un bucle de eventos (event loop). Es ideal para I/O-bound: redes, archivos, operaciones REST, WebSockets. No acelera CPU-bound (usa multiprocessing o C-extensions para eso).

Conceptos clave

• Coroutine: función definida con async def. Se ejecuta hasta un await y cede el control.
• Task: corrutina programada en el loop como unidad ejecutable (asyncio.create_task()).
• Event loop: administra ejecución de tareas y callbacks.
• Future: placeholder de un resultado futuro; las Tasks son Futures.

Ejemplo mínimo

import asyncio

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

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

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

Usa asyncio.run() en el punto de entrada. Evita manipular el loop directamente salvo que tengas razones específicas.

Concurrency patterns

Las formas más comunes de ejecutar múltiples coroutines:

• asyncio.gather: espera a que todas terminen; propaga excepciones (fallo cancela el gather).
• asyncio.as_completed: itera según van terminando (útil para procesar resultados parciales).
• asyncio.wait: más bajo nivel, permite TIMEOUTS y control sobre FIRST_COMPLETED / FIRST_EXCEPTION.

Ejemplo: as_completed

import asyncio

async def worker(i):
    await asyncio.sleep(3 - i)
    return i

async def main():
    tasks = [asyncio.create_task(worker(i)) for i in range(3)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print('got', result)

asyncio.run(main())

Controlar concurrencia: Semaphore

Para evitar sobrecargar recursos remotos o el propio sistema, limita concurrencia con semáforos:

import asyncio
import aiohttp

semaphore = asyncio.Semaphore(10)  # máximo 10 peticiones concurrentes

async def fetch(url):
    async with semaphore:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                return await resp.text()

Mejor aún: reutiliza una única ClientSession en todo el programa para eficiencia (TCP keep-alive).

Timeouts y cancelaciones

Usa asyncio.wait_for o los timeouts nativos de aiohttp para evitar tareas colgadas. Cancela tareas correctamente para liberar recursos.

try:
    result = await asyncio.wait_for(some_coro(), timeout=5.0)
except asyncio.TimeoutError:
    # manejar timeout
    pass

Cuando cancelas una Task, su corrutina recibe asyncio.CancelledError. Asegúrate de manejar limpieza en bloques try/finally dentro de la corrutina.

Exceptions en tareas

Si creas tareas con create_task y no las esperas, las excepciones se perderán salvo que consultes la task o añadas un callback:

def _log_task(task):
    try:
        task.result()
    except Exception as e:
        print('Task error:', e)

t = asyncio.create_task(coro())
t.add_done_callback(_log_task)

Integración con código bloqueante

Para operaciones CPU-bound o librerías bloqueantes: usa run_in_executor (o ThreadPoolExecutor/ProcessPoolExecutor):

import asyncio
import requests

async def blocking_fetch(url):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, requests.get, url)

Preferible: usa librerías asíncronas (aiohttp) cuando estén disponibles.

Depuración

• Activa PYTHONASYNCIODEBUG=1 para detectar bucles bloqueados y corutinas no esperadas.
• Usa asyncio.all_tasks() y asyncio.current_task() para inspección.
• Para profiling, py-spy y otras herramientas funcionan con procesos async.

Buenas prácticas resumidas

• Reutiliza aiohttp.ClientSession y cerrarla correctamente.
• No mezcles asyncio.run y loop.run_forever en la misma app; unifica el punto de entrada.
• Limita concurrencia con semáforos o pools.
• Maneja timeouts y cancels explícitamente.
• Evita bloquear el loop: no sleeps sin await asyncio.sleep(), no I/O sin await.

Errores comunes

• Esperar una coroutine sin await (devuelve coroutine objeto, no resultado).
• No crear tasks cuando quieres que corran en background.
• No cerrar sessions: fugas de sockets.
• Usar demasiados hilos en vez de async para I/O: overhead de threads.

Proyecto de ejemplo: Crawler asíncrono

Objetivo: descargar títulos HTML de una lista de URLs de forma concurrente, con límite de concurrencia, reintentos y logging básico.

Estructura de carpetas

crawler/
├─ pyproject.toml
├─ requirements.txt
└─ crawler/
   ├─ __init__.py
   ├─ main.py
   └─ fetcher.py

requirements.txt

aiohttp>=3.8
async-timeout
beautifulsoup4

crawler/fetcher.py

import asyncio
import aiohttp
from bs4 import BeautifulSoup

class Fetcher:
    def __init__(self, session: aiohttp.ClientSession, concurrency: int = 10):
        self.session = session
        self.semaphore = asyncio.Semaphore(concurrency)

    async def fetch(self, url: str, timeout: float = 10.0, retries: int = 2) -> str:
        for attempt in range(1, retries + 2):
            try:
                async with self.semaphore:
                    async with self.session.get(url, timeout=timeout) as resp:
                        resp.raise_for_status()
                        text = await resp.text()
                        return text
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt <= retries:
                    await asyncio.sleep(0.5 * attempt)  # backoff
                    continue
                raise

    async def fetch_title(self, url: str) -> dict:
        html = await self.fetch(url)
        soup = BeautifulSoup(html, 'html.parser')
        title = soup.title.string.strip() if soup.title and soup.title.string else ''
        return {'url': url, 'title': title}

crawler/main.py

import asyncio
import aiohttp
from crawler.fetcher import Fetcher

URLS = [
    'https://www.python.org',
    'https://www.djangoproject.com',
    'https://pypi.org',
    # añade más URLs
]

async def main(urls):
    timeout = aiohttp.ClientTimeout(total=15)
    connector = aiohttp.TCPConnector(limit=100)  # control de conexiones TCP
    async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
        fetcher = Fetcher(session, concurrency=20)
        tasks = [asyncio.create_task(fetcher.fetch_title(url)) for url in urls]

        results = []
        for coro in asyncio.as_completed(tasks):
            try:
                r = await coro
                print(r)
                results.append(r)
            except Exception as e:
                print('error fetching:', e)
    return results

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

Por qué esta estructura

• Fetcher encapsula lógica HTTP y semáforo: fácil de testear.
• Reutilizar ClientSession reduce latencia por conexiones.
• as_completed permite procesar resultados tan pronto como estén listos y manejar errores por tarea.

Pruebas y observabilidad

• Escribe tests unitarios mockeando aiohttp.ClientSession.get con pytest + pytest-asyncio.
• Métrica importante: latencia P95/P99, tasa de errores por host, sockets abiertos.

Optimización y rendimiento

• Evita crear y cerrar ClientSession por petición.
• Ajusta TCPConnector(limit) y semáforo según capacidad (no pongas valores arbitrarios).
• Usa keep-alive y HTTP/2 si el cliente y servidor lo soportan.
• Si necesitas alto rendimiento en HTTP, considera uvloop (drop-in replacement for asyncio loop) con pruebas: pip install uvloop y asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()).

Errores comunes que evitar en producción

• No controlar timeouts -> tasks colgadas -> memory leak
• Ignorar excepciones en background tasks -> silent failures
• Abrir demasiadas conexiones simultáneas -> sockets agotados

Consejo avanzado: instrumenta las Tasks críticas con métricas custom (latencia, éxito/fracaso) y expónlas para alertas. Advertencia: mezclar operaciones CPU-bound intensas en el mismo proceso async degradará todo el loop; mueva trabajo pesado a ProcessPoolExecutor o servicios separados.

Siguiente paso: implementa circuit breaker por dominio y backpressure para que tu crawler sea robusto ante fallos de servicios externos.