TaskGroup en Python

Una guardería de tareas: espera a todas y cancela el resto ante un error.

import asyncio


async def obtener(nombre, demora):
    await asyncio.sleep(demora)
    return f"{nombre} listo"


async def principal():
    async with asyncio.TaskGroup() as grupo:
        primera = grupo.create_task(obtener("basics", 0.02))
        segunda = grupo.create_task(obtener("traits", 0.01))
    print(primera.result(), segunda.result())

    try:
        async with asyncio.TaskGroup() as grupo:
            grupo.create_task(obtener("ok", 0.01))
            grupo.create_task(asyncio.sleep(0, result=None))
            raise RuntimeError("algo se rompió")
    except* RuntimeError as errores:
        print("capturado", [str(e) for e in errores.exceptions])


asyncio.run(principal())

Cómo funciona

Cada tarea iniciada en el grupo se espera al salir.
Una falla cancela a las hermanas y lanza un ExceptionGroup.
Reemplaza el baile de gather más limpieza manual.

Palabras clave y builtins usados aquí

as
async
await
def
except
for
print
raise
return
str
try
with

El intento, en números

Líneas: 24
Caracteres a escribir: 607
Tokens: 166
Ritmo de tres estrellas: 110 tpm

Al ritmo de tres estrellas de 110 tokens por minuto, este intento toma unos 91 segundos.

Escribe este fragmento

Paso 1 de 4 en Async a fondo; paso 48 de 53 en Python pythónico.

← Anterior Siguiente →