Coordinar resultados y fallos

Consultar tres precios independientes no tiene la misma política que ejecutar tres pasos necesarios para producir un único informe. En el primer caso quizá quieras conservar dos precios aunque el tercero falle. En el segundo, si falta una pieza obligatoria, continuar otras tareas puede desperdiciar recursos o producir un resultado inválido.

Antes de escoger una API, escribe el contrato: qué constituye éxito, qué errores son resultados esperados y qué errores deben detener toda la operación. La diferencia entre gather y TaskGroup no es solo de sintaxis.

asyncio.gather() espera varias operaciones y devuelve una lista con el mismo orden que sus argumentos. Las corrutinas recibidas se programan como tareas. Ese orden no obliga a que terminen una detrás de otra.

import asyncio


async def consultar(nombre, demora):
    await asyncio.sleep(demora)
    return nombre


async def main():
    resultado = await asyncio.gather(
        consultar("primera", 0.1), consultar("segunda", 0.01)
    )
    assert resultado == ["primera", "segunda"]
    print(resultado)


asyncio.run(main())

Su comportamiento por defecto ante una excepción es importante: la propaga al llamador, pero no cancela por ello las otras operaciones. Si el llamador captura el fallo y mantiene vivo el bucle, las demás pueden seguir ejecutándose.

Este programa recoge explícitamente la tarea superviviente para no depender del cierre global del script:

import asyncio


async def fallar():
    await asyncio.sleep(0.02)
    raise ValueError("respuesta inválida")


async def continuar(eventos):
    await asyncio.sleep(0.08)
    eventos.append("terminó la otra consulta")
    return 7


async def main():
    eventos = []
    otra = asyncio.create_task(continuar(eventos))
    try:
        await asyncio.gather(fallar(), otra)
    except ValueError:
        print("gather ya comunicó el error; otra terminada:", otra.done())
    assert not otra.done()
    assert await otra == 7
    print(eventos)


asyncio.run(main())

Si quitaras await otra y main terminara, el cierre de asyncio.run podría cancelar lo pendiente. Atribuir esa cancelación a gather daría una conclusión falsa. Por eso una prueba de semántica debe mantener el entorno vivo el tiempo necesario para observarla.

Cancelar el propio gather mientras está pendiente sí solicita cancelar sus operaciones pendientes. Cancelarlo después de que ya comunicó una excepción no sirve para limpiar automáticamente los trabajos que dejó en marcha. Conserva sus referencias si vas a administrarlos tú.

Un grupo de tareas es un ámbito que posee a sus tareas hijas: no termina de salir hasta que todas han terminado. Si una hija falla con un error distinto de cancelación, TaskGroup solicita cancelar a sus hermanas, espera su limpieza y comunica los errores del grupo.

import asyncio


async def fallar():
    await asyncio.sleep(0.02)
    raise ValueError("respuesta inválida")


async def continuar(eventos):
    try:
        await asyncio.sleep(10)
        eventos.append("trabajo completado")
    finally:
        eventos.append("recurso liberado")


async def main():
    eventos = []
    try:
        async with asyncio.TaskGroup() as grupo:
            grupo.create_task(fallar(), name="consulta-invalida")
            grupo.create_task(continuar(eventos), name="consulta-larga")
    except* ValueError as errores:
        print("Errores de validación:", len(errores.exceptions))
    assert eventos == ["recurso liberado"]
    print(eventos)


asyncio.run(main())

El grupo espera la ejecución del finally. No imprime «trabajo completado». La cancelación es cooperativa: si una hija bloquea el hilo o absorbe la cancelación, la salida puede tardar más de lo previsto. El grupo organiza propiedad; no puede forzar a una función arbitraria a detenerse.

Varias tareas pueden fallar antes de que se complete su cancelación. Python representa esos fallos relacionados mediante ExceptionGroup. except* ValueError selecciona dentro del grupo los errores de ese tipo; los demás siguen propagándose.

No uses except* Exception: pass como solución universal: borrarías tanto un servicio caído como un NameError introducido al programar. En un mismo bloque try tampoco se mezclan cláusulas except y except*; si necesitas distinguir niveles, anida bloques con responsabilidades diferentes.

La cancelación ordinaria de una hija no se trata como un fallo que necesariamente derriba a todas sus hermanas. En cambio, cancelar al dueño del TaskGroup solicita cancelar al conjunto. Esa distinción permite representar cancelaciones esperadas de una operación sin confundirlas con defectos de programación.

No tienes que elegir entre TaskGroup y conservar resultados parciales. Convierte los fallos esperados de cada elemento en un dato dentro de su tarea. Los defectos inesperados siguen siendo excepciones y el grupo los hace visibles.

import asyncio


async def consultar(identificador):
    await asyncio.sleep(0.01)
    if identificador == "roto":
        raise TimeoutError("el servicio no respondió")
    return f"valor-{identificador}"


async def resultado_de(identificador):
    try:
        valor = await consultar(identificador)
        return {"id": identificador, "estado": "ok", "valor": valor}
    except TimeoutError:
        return {"id": identificador, "estado": "timeout", "valor": None}


async def main():
    async with asyncio.TaskGroup() as grupo:
        tareas = [grupo.create_task(resultado_de(i)) for i in ["a", "roto", "b"]]
    resultados = [tarea.result() for tarea in tareas]
    assert [r["estado"] for r in resultados] == ["ok", "timeout", "ok"]
    print(resultados)


asyncio.run(main())

TimeoutError está simulado aquí para centrar la atención en la política. El quinto capítulo lo sustituye por fallos HTTP reales. No capturamos CancelledError, así que cancelar el conjunto sigue funcionando.

gather(..., return_exceptions=True) también permite recibir excepciones junto a valores. Es práctico para operaciones pequeñas o limpieza final, pero obliga a revisar cada posición y evita distinguir tipos de resultado mediante una estructura clara. Una excepción en la lista no debe acabar serializada como si fuera un éxito.

Necesidad	Diseño habitual
Pocas consultas independientes y resultado conjunto sencillo	gather, con política de errores explícita
Varias tareas que pertenecen a una operación	TaskGroup y resultados recogidos al salir
Lote con fallos individuales tolerables	TaskGroup más errores esperados convertidos en resultados
Miles o millones de entradas	Número fijo de workers y cola acotada; no una tarea por entrada
Resultados a medida que llegan	Workers que publican resultados o as_completed para un conjunto ya acotado

as_completed cambia el orden de consumo; no reduce por sí mismo el número de tareas creadas ni cierra las que abandonas. Un orden de salida diferente y un límite de capacidad son problemas distintos.

Sustituye el timeout simulado del último ejemplo por un error de programación, como acceder a una clave inexistente. El programa debe fallar de forma visible. Si termina con una lista aparentemente correcta, tu captura de excepciones está ocultando defectos.

Después cancela al dueño del grupo y comprueba que la cancelación llega a sus hijas. En el próximo capítulo veremos cómo hacerlo sin perder la liberación de recursos.

• TaskGroup y gather, documentación de Python 3.12.
• ExceptionGroup y except*, tutorial oficial.