Programación Asíncrona con asyncio en Python

Programación Asíncrona con asyncio en Python

asyncio es la biblioteca estándar de Python para I/O asíncrono cooperativo. Permite ejecutar miles de operaciones de red simultáneamente en un único hilo, sin el GIL ni la complejidad de los hilos del sistema operativo.

Conceptos clave

Primeros pasos

import asyncio

# La forma más simple de ejecutar una coroutine
async def saludar(nombre: str) -> str:
    await asyncio.sleep(1)   # simula I/O (no bloquea el event loop)
    return f"Hola, {nombre}!"

# Punto de entrada
async def main():
    resultado = await saludar("mundo")
    print(resultado)

asyncio.run(main())  # gestiona el event loop automáticamente

Concurrencia con Tasks y gather

import asyncio
import time

async def fetch_simulado(url: str, delay: float) -> dict:
    """Simula una petición HTTP con delay variable."""
    print(f"  → iniciando {url}")
    await asyncio.sleep(delay)
    print(f"  ✓ completado {url} ({delay:.1f}s)")
    return {"url": url, "status": 200, "size": int(delay * 1000)}

async def main_secuencial():
    """Sin concurrencia — espera cada petición."""
    t0 = time.perf_counter()
    for url, d in [("api/users", 0.8), ("api/posts", 1.2), ("api/comments", 0.5)]:
        await fetch_simulado(url, d)
    print(f"Secuencial: {time.perf_counter()-t0:.2f}s")

async def main_concurrente():
    """Con gather — todas las peticiones en paralelo."""
    t0 = time.perf_counter()
    resultados = await asyncio.gather(
        fetch_simulado("api/users",    0.8),
        fetch_simulado("api/posts",    1.2),
        fetch_simulado("api/comments", 0.5),
    )
    print(f"Concurrente: {time.perf_counter()-t0:.2f}s")
    return resultados

asyncio.run(main_secuencial())   # ~2.5s
asyncio.run(main_concurrente())  # ~1.2s  (limitado por la más lenta)

Tasks — control granular

import asyncio

async def tarea_con_progreso(nombre: str, pasos: int):
    for i in range(1, pasos + 1):
        await asyncio.sleep(0.1)
        print(f"{nombre}: paso {i}/{pasos}")
    return f"{nombre} completada"

async def main():
    # Crear tasks (se inician inmediatamente)
    task_a = asyncio.create_task(tarea_con_progreso("A", 3), name="tarea-A")
    task_b = asyncio.create_task(tarea_con_progreso("B", 5), name="tarea-B")

    # Esperar a que task_a termine primero
    resultado_a = await task_a
    print(f"Task A: {resultado_a}")

    # task_b sigue corriendo — esperamos también
    resultado_b = await task_b
    print(f"Task B: {resultado_b}")

asyncio.run(main())

gather vs TaskGroup (Python 3.11+)

# gather — comportamiento ante errores configurable
resultados = await asyncio.gather(
    coro1(), coro2(), coro3(),
    return_exceptions=True,  # no aborta si una falla; devuelve la excepción
)

# TaskGroup — cancela todo si una tarea falla (recomendado para 3.11+)
async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(coro1())
        t2 = tg.create_task(coro2())
    # aquí ambas tareas han terminado (o se lanzó ExceptionGroup)
    print(t1.result(), t2.result())

Timeouts

import asyncio

async def operacion_lenta():
    await asyncio.sleep(5)
    return "datos"

async def main():
    # Python 3.11+: asyncio.timeout (context manager)
    try:
        async with asyncio.timeout(2.0):
            resultado = await operacion_lenta()
    except TimeoutError:
        print("Operación cancelada por timeout")

    # Alternativa compatible con 3.9+
    try:
        resultado = await asyncio.wait_for(operacion_lenta(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Timeout (wait_for)")

asyncio.run(main())

Semáforos y control de concurrencia

import asyncio
import aiohttp  # pip install aiohttp

MAX_CONCURRENTES = 10

async def fetch_url(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> dict:
    async with sem:  # limita a MAX_CONCURRENTES peticiones simultáneas
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            data = await resp.json()
            return {"url": url, "status": resp.status, "data": data}

async def fetch_many(urls: list[str]) -> list[dict]:
    sem = asyncio.Semaphore(MAX_CONCURRENTES)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url, sem) for url in urls]
        resultados = await asyncio.gather(*tasks, return_exceptions=True)

    # Separar éxitos de errores
    exitos  = [r for r in resultados if isinstance(r, dict)]
    errores = [r for r in resultados if isinstance(r, Exception)]
    print(f"OK: {len(exitos)} | Errores: {len(errores)}")
    return exitos

# Uso
urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 51)]
resultados = asyncio.run(fetch_many(urls))

Patrón productor-consumidor con Queue

import asyncio
import random

async def productor(queue: asyncio.Queue, n: int, pid: int):
    for i in range(n):
        item = f"item-{pid}-{i}"
        await queue.put(item)
        print(f"[P{pid}] produjo {item} (queue size: {queue.qsize()})")
        await asyncio.sleep(random.uniform(0.05, 0.15))
    print(f"[P{pid}] terminado")

async def consumidor(queue: asyncio.Queue, cid: int):
    while True:
        try:
            item = await asyncio.wait_for(queue.get(), timeout=1.0)
            print(f"  [C{cid}] procesando {item}")
            await asyncio.sleep(random.uniform(0.1, 0.3))  # simula trabajo
            queue.task_done()
        except asyncio.TimeoutError:
            print(f"  [C{cid}] sin trabajo, saliendo")
            break

async def pipeline():
    queue = asyncio.Queue(maxsize=20)  # bufferiza hasta 20 items

    # Lanzar productores y consumidores concurrentemente
    async with asyncio.TaskGroup() as tg:
        for i in range(3):
            tg.create_task(productor(queue, n=5, pid=i))
        for j in range(4):
            tg.create_task(consumidor(queue, cid=j))

asyncio.run(pipeline())

API HTTP asíncrona con aiohttp

import asyncio
import aiohttp
from typing import Any

async def api_client(base_url: str) -> list[Any]:
    """Descarga todos los posts con paginación automática."""
    resultados = []
    pagina = 1

    async with aiohttp.ClientSession(
        headers={"Accept": "application/json"},
        timeout=aiohttp.ClientTimeout(total=30),
    ) as session:
        while True:
            async with session.get(f"{base_url}?_page={pagina}&_limit=10") as resp:
                if resp.status != 200:
                    break
                datos = await resp.json()
                if not datos:
                    break
                resultados.extend(datos)
                pagina += 1

    return resultados

async def main():
    posts = await api_client("https://jsonplaceholder.typicode.com/posts")
    print(f"Total posts descargados: {len(posts)}")
    for post in posts[:3]:
        print(f"  [{post['id']}] {post['title'][:50]}...")

asyncio.run(main())

asyncio con streams (servidor TCP)

import asyncio

async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    addr = writer.get_extra_info('peername')
    print(f"Conexión de {addr}")

    while True:
        data = await reader.read(1024)
        if not data:
            break
        mensaje = data.decode().strip()
        respuesta = f"Echo: {mensaje}\n"
        writer.write(respuesta.encode())
        await writer.drain()

    print(f"Cerrando {addr}")
    writer.close()
    await writer.wait_closed()

async def servidor():
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8888)
    addr = server.sockets[0].getsockname()
    print(f"Servidor escuchando en {addr}")
    async with server:
        await server.serve_forever()

# asyncio.run(servidor())

Debugging y monitoreo

import asyncio
import logging

# Activar modo debug (detecta coroutines no esperadas, operaciones lentas)
asyncio.run(main(), debug=True)

# Log de operaciones lentas (umbral por defecto: 100ms)
logging.basicConfig(level=logging.DEBUG)

# Ver todas las tasks activas desde dentro de una coroutine
async def inspeccionar():
    tasks = asyncio.all_tasks()
    for t in tasks:
        print(f"  Task: {t.get_name()!r} — done={t.done()}")

Buenas prácticas

• asyncio.run(main()) es el punto de entrada correcto en scripts; no llames a loop.run_until_complete() manualmente (API antigua).
• No mezcles código bloqueante con await — time.sleep(), operaciones de archivo síncronas o código CPU-intensivo bloquean el event loop. Usa asyncio.to_thread() o loop.run_in_executor() para delegar a un thread pool.
• asyncio.gather(return_exceptions=True) para no abortar toda la operación si una tarea falla; procesa los resultados separando diccionarios de excepciones.
• Usa asyncio.TaskGroup (3.11+) cuando quieras que un fallo en cualquier tarea cancele automáticamente las demás — es más seguro que gather sin manejo de errores.
• Controla la concurrencia con asyncio.Semaphore al hacer scraping o llamadas a APIs con rate limiting.
• Cierra siempre las sesiones de aiohttp con async with aiohttp.ClientSession() — evita warnings de conexiones sin cerrar.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF