asyncio.TaskGroup en Python: Concurrencia Estructurada

Crear tareas es sencillo; asegurar que todas terminen y sus excepciones sean observadas es más difícil. asyncio.TaskGroup da un ciclo de vida visible a tareas relacionadas desde Python 3.11.

Revisa async y await antes de usar concurrencia.

import asyncio

async def buscar(nombre: str, demora: float) -> str:
    await asyncio.sleep(demora)
    return f"{nombre}: ok"

async def main():
    async with asyncio.TaskGroup() as grupo:
        usuarios = grupo.create_task(buscar("usuarios", 0.2))
        pedidos = grupo.create_task(buscar("pedidos", 0.1))
    print(usuarios.result(), pedidos.result())

El contexto espera ambas tareas. Si una falla, cancela las restantes y puede producir ExceptionGroup, tratable con except*. No ocultes CancelledError. La documentación oficial detalla esta semántica.

gather() sigue siendo útil, pero TaskGroup es más seguro cuando las tareas forman una unidad. Ninguno acelera CPU; consulta threads y multiprocessing.

Usa timeouts, libera recursos en finally, limita concurrencia hacia APIs y nombra tareas cuando ayude a observarlas. La concurrencia estructurada reduce tareas huérfanas.

Guardar tareas y recuperar resultados

TaskGroup.create_task() devuelve un objeto Task. Conviene conservarlo cuando hay que relacionar cada resultado con su entrada:

async def consultar_todos(ids: list[int]) -> dict[int, dict]:
    tareas: dict[int, asyncio.Task[dict]] = {}

    async with asyncio.TaskGroup() as grupo:
        for cliente_id in ids:
            tareas[cliente_id] = grupo.create_task(
                consultar_cliente(cliente_id),
                name=f"cliente-{cliente_id}",
            )

    return {
        cliente_id: tarea.result()
        for cliente_id, tarea in tareas.items()
    }

Lee result() después de una salida normal del contexto. Dentro del bloque, la tarea todavía puede estar ejecutándose. Si una consulta falla, el flujo no alcanza el return: el grupo cierra las tareas restantes y propaga el fallo. Los nombres no cambian la planificación, pero ayudan a interpretar logs y diagnósticos.

Una coroutine puede añadir trabajo al mismo grupo mientras este siga activo. Eso permite construir árboles de tareas descubiertos dinámicamente. Después de que termina la última tarea y se cierra el contexto, ya no se pueden añadir más.

La cancelación necesita cooperación

Asyncio no termina una coroutine por la fuerza. La cancelación produce CancelledError en un punto de suspensión, como await. El código debe liberar recursos y permitir que la excepción continúe:

async def consumir():
    conexion = await abrir_conexion()
    try:
        while True:
            elemento = await recibir(conexion)
            await procesar(elemento)
    finally:
        await conexion.close()

Evita capturar BaseException. Si necesitas capturar CancelledError para registrar estado o limpiar recursos, termina esa labor y vuelve a lanzarla con raise. Los grupos y asyncio.timeout() usan cancelación internamente. Ocultarla puede retrasar el cierre y romper las garantías del bloque.

KeyboardInterrupt y SystemExit tienen un tratamiento especial. El grupo cancela y espera a sus hijos, pero vuelve a lanzar la excepción original en lugar de incluirla en un grupo. Los fallos comunes sí pueden combinarse cuando varias tareas fallan durante el cierre.

Tratar ExceptionGroup de forma selectiva

La sintaxis except* selecciona partes de un ExceptionGroup según su tipo:

try:
    async with asyncio.TaskGroup() as grupo:
        for ruta in rutas:
            grupo.create_task(importar(ruta))
except* ArchivoInvalido as errores:
    for error in errores.exceptions:
        registrar_rechazo(error)
except* TimeoutError as errores:
    for error in errores.exceptions:
        registrar_timeout(error)

Tratar un tipo no elimina fallos no relacionados. Las excepciones que no coinciden siguen hacia la capa superior. Así se evita que un defecto de programación parezca un resultado correcto. A menudo conviene añadir contexto cerca de cada operación y dejar que el grupo llegue a una capa capaz de decidir si reintenta, devuelve un error o detiene el proceso.

No bases reglas de negocio en el orden de las excepciones. El orden de finalización cambia con la red, el sistema operativo y la carga. Si cada elemento puede fallar sin invalidar los demás, captura el error esperado dentro de la propia tarea y devuelve un valor explícito de éxito o fallo. El grupo considerará terminada esa tarea y no cancelará a sus hermanas.

Un timeout para toda la operación

Un timeout exterior establece un presupuesto para el trabajo completo:

async def cargar_panel() -> tuple[dict, list]:
    async with asyncio.timeout(2.0):
        async with asyncio.TaskGroup() as grupo:
            resumen = grupo.create_task(buscar_resumen())
            alertas = grupo.create_task(buscar_alertas())

    return resumen.result(), alertas.result()

Al vencer el plazo, el contexto de timeout cancela la tarea actual. El TaskGroup cierra entonces a sus hijos antes de que TimeoutError llegue al llamador. Un timeout independiente dentro de cada tarea expresa otra política: una dependencia lenta falla y, si la excepción escapa, normalmente cancela a sus hermanas.

TaskGroup tampoco limita por sí solo la concurrencia. Crear diez mil tareas que esperan un semáforo mantiene diez mil objetos y sus argumentos en memoria. Para flujos grandes, usa una cantidad fija de workers con asyncio.Queue, procesa páginas o aplica otra forma de contrapresión.

TaskGroup frente a gather

asyncio.gather() es conciso para una colección fija de awaitables cuando interesa recibir resultados en el orden de entrada. Con return_exceptions=True, puede representar fallos como valores, algo apropiado para operaciones realmente independientes. En ese caso hay que inspeccionar cada posición; ignorar una excepción devuelta crea un falso éxito.

TaskGroup no devuelve una lista ni ofrece un equivalente directo a return_exceptions=True. A cambio, vincula creación, espera y cancelación a un ámbito visible. Es una opción clara cuando las subtareas forman una unidad, por ejemplo las partes obligatorias de una respuesta.

No toda operación asíncrona debe ejecutarse en paralelo. Si B necesita el resultado de A, dos await secuenciales comunican mejor la dependencia. La concurrencia también puede agotar pools de conexión, cuotas de API, descriptores y memoria. Define límites y mide latencia y capacidad.

Probar fallos y limpieza

Una prueba valiosa fuerza el fallo de una tarea y verifica que su hermana ejecuta la limpieza. Coordina el escenario con asyncio.Event en vez de depender solo de pausas arbitrarias. Prueba además el contrato público: qué excepción recibe el llamador, cuáles efectos parciales son aceptables y si repetir la operación es seguro.

TaskGroup forma parte de la biblioteca estándar desde Python 3.11. Si un paquete admite versiones anteriores, declara esa restricción y elige conscientemente una alternativa compatible. En aplicaciones actuales, un bloque estructurado suele ser más fácil de revisar que llamadas dispersas a create_task() sin propietario claro.

Revisión antes de producción

Antes de publicar, identifica qué tareas son obligatorias y qué fallos esperados deberían convertirse en valores en vez de excepciones. Coloca el timeout alrededor del presupuesto real del servicio y confirma que cada coroutine libera conexiones, locks y archivos temporales al cancelarse.

Revisa capacidad además de corrección. Calcula el máximo de llamadas simultáneas producido por grupos anidados y compáralo con pools y cuotas externas. Si la cantidad de entradas no tiene límite, introduce cola, paginación o lotes antes de crear las tareas.

Haz visible la propiedad. Una tarea creada dentro del grupo debe terminar allí; un trabajo de fondo que sobreviva a una petición requiere otro ciclo de vida supervisado. En pruebas cubre finalización normal, fallo de un hijo, varios fallos durante limpieza, cancelación del llamador y vencimiento del plazo. Esos casos encuentran problemas que una demostración exitosa no revela.

Incluye además identificadores estables de la operación en los logs. Los nombres de tareas ayudan a depurar, pero no sustituyen contexto de negocio ni métricas sobre duración, errores y cancelaciones.