Tasks de asyncio — Ejecución concurrente con gather, Task y Queue

Sobre la base de async def y await, este artículo cubre cómo ejecutar varias corrutinas en concurrencia. asyncio.gather para "dispara todo y espera todo", asyncio.create_task para "dispara ahora, espera después" y asyncio.Queue para productor / consumidor — estos tres cubren prácticamente cualquier patrón async que vayas a escribir en proyectos reales.

Sobre la ejecución de código aquí

async / await es todo cuestión de tiempos, pero el runner de este sitio acumula la salida de print y la muestra toda de golpe cuando termina el script. La salida en tiempo real y la sensación del tiempo transcurrido no coincidirán con un entorno Python real. El artículo usa diagramas para dejar claro el comportamiento interno, pero si quieres ver el flujo de print en vivo o sentir los tiempos reales, ejecútalo en tu entorno Python local con asyncio.run.

asyncio.gather — ejecutar varias corrutinas en concurrencia

"Dispara varias APIs en concurrencia y solo avanza cuando estén todas las respuestas" — un caso de uso clásico de async. El código síncrono suma los tiempos de respuesta, pero gather termina en el tiempo de la más lenta.

asyncio.gather(coro1, coro2, ...) arranca a la vez todas las corrutinas que le pases y devuelve una lista de resultados cuando todas terminan. La lista regresa en el orden en que las pasaste — no en orden de finalización — así que la entrada y la salida quedan alineadas.

import asyncio

async def fetch(name):
    await asyncio.sleep(1)        # Simula una llamada API con espera de 1 seg
    return f"{name} done"

# Dispara 3 en paralelo → todos los resultados en 1 seg (vs 3 seg secuencial)
results = await asyncio.gather(
    fetch("A"),
    fetch("B"),
    fetch("C"),
)
print(results)                    # ['A done', 'B done', 'C done']  ← orden de entrada

Qué pasa con las excepciones

Por defecto, si alguna corrutina dentro de gather lanza, toda la llamada lanza y aborta. Para recolectar las excepciones en su lugar, pasa asyncio.gather(..., return_exceptions=True) — los objetos de excepción regresan entonces como elementos de la lista, así puedes comprobar los tipos a posteriori. Útil cuando quieres llamar a varias APIs y tolerar fallos parciales.

Ejecuta 3 corrutinas en concurrencia con asyncio.gather y confirma que el orden de finalización puede diferir pero el valor de retorno se mantiene en orden de entrada. Cada tarea espera una cantidad de tiempo distinta para que veas cómo el orden de finalización (B → A → C) y el orden de retorno ([A, B, C]) se separan.

① Añade import asyncio.

② Define async def task(name, secs): — await asyncio.sleep(secs) y luego return f"{name} done".

③ Ejecuta con await asyncio.gather(task("A", 0.3), task("B", 0.1), task("C", 0.5)) y guarda en results.

④ Imprime print("results:", results).

⑤ Imprime print("count:", len(results)).

(Ejecuta con éxito y aparecerá la explicación.)

create_task y wait_for — dispara ahora, espera después, con timeout

asyncio.create_task(coro) envuelve una corrutina en un objeto "Task" y la arranca de inmediato. A diferencia del "dispara todo y espera todo" de gather, este es el patrón clásico async de "dispara ahora, haz otro trabajo y luego await task para recoger el resultado más tarde".

asyncio.wait_for(awaitable, timeout=N) es una red de seguridad: "lanza TimeoutError si no termina en N segundos". Es estándar combinarlo con Tasks como salvaguarda cuando una Web API no responde.

import asyncio

async def slow_api():
    await asyncio.sleep(2)
    return "response"

# Dispárala con create_task (la Task arranca corriendo de inmediato)
task = asyncio.create_task(slow_api())

# Mientras la Task corre en segundo plano, puede pasar otro trabajo
print("task fired, doing other work...")

# Recoge el resultado con await cuando lo necesites
result = await task
print(result)                       # response

# wait_for añade un timeout (rendirse tras 1 seg)
try:
    result = await asyncio.wait_for(slow_api(), timeout=1.0)
except asyncio.TimeoutError:
    print("timeout!")               # respuesta de 2 seg vs presupuesto de 1 seg → aquí

Tres caminos a "done" — return / excepción / cancel

Hay 3 caminos a done: (1) finalización normal — return produjo un valor, (2) excepción — algo lanzó dentro, (3) cancel — task.cancel() la interrumpió. task.done() devuelve True para los tres caminos, y task.exception() extrae la excepción si la hay.

Métodos útiles del objeto Task

El objeto Task que devuelve create_task admite operaciones útiles: task.cancel() para interrumpir, task.done() para comprobar finalización, task.result() para obtener el resultado de una Task terminada (lanza si no está done) y task.exception() para obtener cualquier excepción lanzada. Útil para controlar trabajo en segundo plano de larga duración.

Dispara dos Tasks con create_task, haz otro trabajo en medio y luego recoge sus resultados.

① Añade import asyncio.

② Define async def task(name): — await asyncio.sleep(0) para cambiar fuera, y luego return f"{name} done".

③ Usa asyncio.create_task(task("A")) y asyncio.create_task(task("B")) para disparar 2 Tasks, guardándolas en t1 y t2.

④ Mientras corren las Tasks, imprime tasks fired.

⑤ Usa await t1 y await t2 para recoger cada resultado e imprime como A: ◯ / B: ◯.

Usa asyncio.wait_for para fijar un timeout y prueba los casos dentro del presupuesto y por encima del presupuesto.

① Añade import asyncio.

② Define async def slow_task(): — await asyncio.sleep(0.5) para esperar 0.5 seg, y luego return "response done".

③ await asyncio.wait_for(slow_task(), timeout=1.0) — termina dentro de 1 seg, así que tiene éxito. Imprime como success: ◯.

④ try: / except asyncio.TimeoutError: alrededor de await asyncio.wait_for(slow_task(), timeout=0.1) — 0.1 seg no es suficiente, así que se dispara TimeoutError. Captúralo e imprime timeout!.

asyncio.Queue — productor / consumidor

"Quiero que una corrutina alimente valores y otra los consuma" — un patrón frecuente en scraping, procesamiento de jobs, manejo de streams y otras situaciones donde dos bucles a velocidades distintas necesitan engranar.

asyncio.Queue es una cola async para pasar valores entre corrutinas (un FIFO = First In First Out — los valores salen en el orden en que los pones). Usa await queue.put(value) para insertar y await queue.get() para sacar — y cuando la cola está vacía / llena, automáticamente cambia a otra tarea y espera, manteniendo todo simple.

import asyncio

queue = asyncio.Queue()

# Pone valores
await queue.put("item-1")
await queue.put("item-2")

# Los saca (FIFO = first in, first out)
print(await queue.get())            # item-1
print(await queue.get())            # item-2

# get() en una cola vacía cambia a otra tarea y espera por un valor
# print(await queue.get())          # ← se pausa aquí hasta que alguien haga put

Para detener limpiamente, usa un sentinel

Del lado del consumidor, while True: item = await queue.get() espera para siempre a que llegue algo. El patrón es que el productor empuje un marcador de terminación al final (típicamente None, o un sentinel = un objeto guardián dedicado que distingues de los datos reales). El consumidor sale del bucle en cuanto ve el marcador.

Prueba put y get dentro de una sola corrutina para confirmar los fundamentos de Queue y FIFO (first in, first out).

① Añade import asyncio y crea una asyncio.Queue vacía.

② await queue.put("a"), "b", "c" — inserta 3 valores en orden.

③ Llama a await queue.get() 3 veces y reúne los valores en una lista, luego imprime como pulled: ◯.

Ejecutar productor / consumidor en concurrencia con gather

Queue da realmente sus frutos cuando varias corrutinas se pasan valores de un lado a otro. Separa "alimentar" y "consumir" en corrutinas distintas y ejecútalas en concurrencia con gather — await put / await get actúan como puntos de cambio, y las dos mitades engranan de forma natural.

import asyncio

async def producer(queue):
    for i in range(3):
        await queue.put(f"item-{i}")
    await queue.put(None)           # marcador de terminación

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:            # rompe con el marcador de terminación
            break
        print(f"processing: {item}")

queue = asyncio.Queue()
await asyncio.gather(producer(queue), consumer(queue))
# Salida:
# processing: item-0
# processing: item-1
# processing: item-2

Construye una configuración en la que el productor pone 3 ítems y el consumidor los recolecta en una lista. Usa None como marcador de terminación.

① Añade import asyncio, crea una asyncio.Queue vacía y una lista vacía results.

② Define async def producer(queue): — pon f"item-{i}" 3 veces, y luego pon None como marcador de terminación.

③ Define async def consumer(queue, results): — while True: llama a get; si None entonces break, si no añade a results.

④ Ejecuta con asyncio.gather(producer(queue), consumer(queue, results)) e imprime como count: ◯ / first: ◯ / last: ◯.