Python asyncio: async y await

El módulo asyncio de Python te permite escribir código concurrente en un solo hilo usando las palabras clave async y await. En lugar de bloquearse mientras espera respuestas de red o lecturas de archivos, un programa asyncio suspende la tarea en espera y cambia inmediatamente a otro trabajo — retomando cuando el resultado está listo. Esto lo convierte en la herramienta adecuada para programas ligados a I/O, como scrapers web, clientes API y servidores de chat.

Este capítulo cubre:

• Qué son las funciones async (corrutinas) y cómo difieren de las funciones regulares
• El bucle de eventos y cómo asyncio planifica el trabajo
• Esperar resultados, ejecutar tareas de forma concurrente con asyncio.gather y asyncio.create_task
• Manejar excepciones y timeouts dentro de código async
• El asyncio.Queue para patrones productor-consumidor
• Cuándo usar asyncio y cuándo recurrir a threading en su lugar

Por qué existe asyncio

Considera un programa que llama a dos APIs una tras otra:

import time

def fetch(name, delay):
    time.sleep(delay)          # blocks the whole program
    return f'data from {name}'

start = time.perf_counter()
r1 = fetch('API A', 1)
r2 = fetch('API B', 1)
print(f'Done in {time.perf_counter() - start:.1f}s')
# Done in 2.0s

Ambas llamadas se ejecutan de forma secuencial — 2 segundos en total, aunque cada llamada solo necesita 1 segundo de espera. Con asyncio el programa pausa fetch('API A', ...) mientras espera, inicia fetch('API B', ...) de inmediato, y ambas terminan en aproximadamente 1 segundo:

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)   # suspends only this coroutine
    return f'data from {name}'

async def main():
    start = time.perf_counter()
    r1, r2 = await asyncio.gather(fetch('API A', 1), fetch('API B', 1))
    print(f'Done in {time.perf_counter() - start:.1f}s')
    # Done in 1.0s

asyncio.run(main())

Corrutinas: async def y await

Una función definida con async def se llama función corrutina. Llamarla no ejecuta el cuerpo de inmediato — devuelve un objeto corrutina que debe ser manejado por el bucle de eventos.

async def greet(name):
    print(f'Hello, {name}!')

# Calling it returns a coroutine object, nothing is printed yet
coro = greet('World')
print(type(coro))   # <class 'coroutine'>

# Run it properly
import asyncio
asyncio.run(greet('World'))
# Hello, World!

Dentro de una corrutina, await suspende la ejecución hasta que el awaitable (otra corrutina, una Task o un Future) produce un resultado. El bucle de eventos puede ejecutar otras corrutinas mientras una está suspendida.

import asyncio

async def step_one():
    print('Step 1: start')
    await asyncio.sleep(1)     # suspend for 1 second
    print('Step 1: end')
    return 'result-1'

async def main():
    value = await step_one()   # wait for step_one to finish
    print(value)

asyncio.run(main())
# Step 1: start
# Step 1: end
# result-1

Qué puedes esperar con await

• Otra corrutina async def
• Un asyncio.Task (creado con asyncio.create_task)
• Un asyncio.Future
• Cualquier objeto con un método __await__

No puedes usar await fuera de una función async def.

El bucle de eventos

El bucle de eventos es el planificador de asyncio. Mantiene una cola de corrutinas y tareas, ejecuta cada una hasta que llega a un await, y luego cambia al siguiente elemento listo. Generalmente hay un bucle de eventos por hilo.

asyncio.run(coro) es el punto de entrada estándar para los programas asyncio. Crea un nuevo bucle de eventos, ejecuta la corrutina dada hasta su finalización, cierra el bucle y devuelve el resultado:

import asyncio

async def compute():
    await asyncio.sleep(0)   # yield control once
    return 6 * 7

result = asyncio.run(compute())
print(result)   # 42

Para la mayoría de las aplicaciones nunca necesitas gestionar el bucle directamente — asyncio.run se encarga de la creación y el cierre.

Ejecutar tareas de forma concurrente

asyncio.gather

asyncio.gather(*coroutines) planifica todas las corrutinas proporcionadas para que se ejecuten de forma concurrente y devuelve sus resultados en el mismo orden:

import asyncio

async def fetch_data(name, delay):
    print(f'Start fetching {name}')
    await asyncio.sleep(delay)
    print(f'Done fetching {name}')
    return f'data from {name}'

async def main():
    results = await asyncio.gather(
        fetch_data('API A', 1),
        fetch_data('API B', 2),
        fetch_data('API C', 1),
    )
    print(results)

asyncio.run(main())
# Start fetching API A
# Start fetching API B
# Start fetching API C
# Done fetching API A
# Done fetching API C
# Done fetching API B
# ['data from API A', 'data from API B', 'data from API C']

Las tres corrutinas comienzan de inmediato. El tiempo total transcurrido coincide con la corrutina más lenta (2 s), no con la suma (4 s).

asyncio.create_task

asyncio.create_task(coro) envuelve una corrutina en una Task y la planifica para que se ejecute pronto. A diferencia de gather, crear una tarea la lanza en segundo plano mientras la corrutina actual sigue ejecutándose:

import asyncio

async def background_job(name, delay):
    print(f'{name}: start')
    await asyncio.sleep(delay)
    print(f'{name}: end')
    return f'{name} done'

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

    # Both tasks are already scheduled; await collects their results
    result1 = await t1
    result2 = await t2
    print(result1, result2)

asyncio.run(main())
# Task A: start
# Task B: start
# Task A: end
# Task B: end
# Task A done Task B done

Usa create_task cuando quieras que una tarea empiece de inmediato y planifiques recoger su resultado (o cancelarla) más adelante. Usa gather cuando quieras lanzar un grupo fijo de corrutinas y esperar a todas ellas juntas.

Salida intercalada

Una forma útil de ver el bucle de eventos en acción es observar cómo se intercalan las tareas:

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f'{name}: {i}')
        await asyncio.sleep(1)
    print(f'{name}: done!')

async def main():
    await asyncio.gather(
        count_down('Task A', 3),
        count_down('Task B', 2),
    )

asyncio.run(main())
# Task A: 3
# Task B: 2
# Task A: 2
# Task B: 1
# Task A: 1
# Task B: done!
# Task A: done!

Ambas tareas comparten un hilo; el bucle de eventos alterna entre ellas en cada await asyncio.sleep(1).

Manejo de excepciones

Las excepciones lanzadas dentro de una corrutina se propagan a través de await igual que en código síncrono. Usa un bloque try/except normal:

import asyncio

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError('something went wrong')

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f'Caught: {e}')

asyncio.run(main())
# Caught: something went wrong

Cuando se usa asyncio.gather, si una corrutina lanza una excepción las otras no se cancelan por defecto, pero la excepción se vuelve a lanzar cuando haces await en la llamada a gather. Pasa return_exceptions=True para recoger las excepciones como valores de retorno en su lugar:

import asyncio

async def good():
    return 'ok'

async def bad():
    raise RuntimeError('oops')

async def main():
    results = await asyncio.gather(good(), bad(), return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f'Error: {r}')
        else:
            print(f'Result: {r}')

asyncio.run(main())
# Result: ok
# Error: oops

Timeouts con asyncio.wait_for

asyncio.wait_for(coro, timeout) ejecuta una corrutina y la cancela si no termina dentro del número de segundos dado, lanzando asyncio.TimeoutError:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return 42

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print('Timed out — operation cancelled')

asyncio.run(main())
# Timed out — operation cancelled

Esto es importante para el código de red en producción donde un servidor bloqueado de otro modo mantendría una tarea bloqueada indefinidamente.

asyncio.Queue para patrones productor-consumidor

asyncio.Queue es una cola segura para hilos y con soporte async. Es ideal para desacoplar productores (código que genera trabajo) de consumidores (código que lo procesa):

import asyncio

async def producer(queue):
    for i in range(1, 4):
        print(f'Produced item {i}')
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)   # sentinel to signal consumers to stop

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed item {item}')

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
# Produced item 1
# Consumed item 1
# Produced item 2
# Consumed item 2
# Produced item 3
# Consumed item 3

Para múltiples consumidores, usa queue.task_done() y queue.join() para saber cuándo se han procesado todos los elementos.

asyncio vs threading

Tanto asyncio como el módulo threading de Python permiten que el trabajo avance de forma concurrente, pero lo hacen de manera diferente:

Usa asyncio cuando controles el código de I/O y puedas usar bibliotecas compatibles con async (por ejemplo, aiohttp, asyncpg). Usa threading cuando dependas de bibliotecas bloqueantes de terceros que no puedan hacerse async.

Para verdadero paralelismo en CPU, recurre a multiprocessing o concurrent.futures.ProcessPoolExecutor en su lugar.

Errores comunes

Olvidar await: Llamar a una función async sin await devuelve un objeto corrutina y no hace nada. Python emite un RuntimeWarning: coroutine '...' was never awaited para ayudar a detectar esto.

async def main():
    asyncio.sleep(1)   # BUG: returns a coroutine, does not sleep
    await asyncio.sleep(1)   # correct

Bloquear el bucle de eventos: Ejecutar código síncrono lento (un bucle ajustado, una llamada de red bloqueante, time.sleep) dentro de una corrutina congela todo el bucle de eventos. Envuelve las llamadas bloqueantes con asyncio.to_thread (Python 3.9+) para ejecutarlas en un pool de hilos sin bloquear:

import asyncio
import time

def blocking_task():
    time.sleep(2)   # simulates a slow blocking operation
    return 'done'

async def main():
    result = await asyncio.to_thread(blocking_task)
    print(result)

asyncio.run(main())
# done

Usar asyncio.run dentro de un bucle en ejecución: Los notebooks de Jupyter ya ejecutan un bucle de eventos. Usa await coro directamente en las celdas del notebook, o instala nest_asyncio para permitir bucles anidados.

Resumen de referencia rápida

Práctica