Crear tareas con asyncio.create_task() es sencillo, pero coordinar su ciclo de vida puede ser difícil. Una tarea puede fallar mientras otras continúan, las referencias pueden perderse y el cierre puede ocurrir antes de completar el cleanup. asyncio.TaskGroup coloca tareas relacionadas dentro de un ámbito estructurado: al salir del contexto, todas han terminado, fueron canceladas o aportaron un error.
Esta guía cubre creación de tareas, resultados, cancelación, ExceptionGroup, comparación con asyncio.gather(), grupos anidados, timeouts, límites de capacidad y limpieza confiable.
El problema de las tareas sueltas
Una tarea creada con create_task() empieza de forma independiente. El llamador debe conservar la referencia, esperarla y manejar sus errores.
import asyncio
async def trabajo(nombre: str) -> str:
await asyncio.sleep(0.2)
return nombre.upper()
async def main():
tareas = [
asyncio.create_task(trabajo("a")),
asyncio.create_task(trabajo("b")),
]
resultados = await asyncio.gather(*tareas)
print(resultados)
asyncio.run(main())El patrón funciona, pero una tarea puede sobrevivir a la operación lógica que la creó. TaskGroup hace explícita la propiedad.
Primer TaskGroup
async def main():
async with asyncio.TaskGroup() as grupo:
tarea_a = grupo.create_task(trabajo("a"))
tarea_b = grupo.create_task(trabajo("b"))
print(tarea_a.result())
print(tarea_b.result())El contexto no termina hasta que todas las tareas hijas finalizan. Después del bloque, los resultados están disponibles y no queda trabajo pendiente propiedad del grupo.
Concurrencia estructurada
La concurrencia estructurada significa que las tareas hijas viven dentro de un ámbito claro. El código que las crea también es responsable de su finalización. Esto simplifica el razonamiento sobre recursos, cancelación y errores.
El concepto amplía la guía de asyncio en Python. Las coroutines aportan concurrencia cooperativa; TaskGroup aporta estructura de ciclo de vida.
Cuando una tarea falla
Si una hija lanza una excepción distinta de cancelación, TaskGroup normalmente cancela las hermanas que siguen activas. Después de detenerlas, las fallas se propagan como ExceptionGroup.
async def fallo_rapido():
await asyncio.sleep(0.1)
raise ValueError("datos inválidos")
async def trabajo_lento():
try:
await asyncio.sleep(10)
finally:
print("cleanup lento")
async def main():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(fallo_rapido())
grupo.create_task(trabajo_lento())La tarea lenta recibe cancelación, ejecuta su finally y solo entonces el grupo propaga los errores.
Tratar ExceptionGroup con except*
async def main():
try:
async with asyncio.TaskGroup() as grupo:
grupo.create_task(fallo_rapido())
grupo.create_task(trabajo_lento())
except* ValueError as errores:
for error in errores.exceptions:
print("error de validación:", error)except* selecciona las excepciones compatibles dentro del grupo. Los tipos no tratados continúan propagándose.
Recoger resultados
TaskGroup.create_task() devuelve un objeto Task. Conserva las referencias cuando necesites los resultados individuales.
async def consultar(item_id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": item_id}
async def cargar(ids: list[int]) -> list[dict]:
tareas: list[asyncio.Task[dict]] = []
async with asyncio.TaskGroup() as grupo:
for item_id in ids:
tareas.append(grupo.create_task(consultar(item_id)))
return [tarea.result() for tarea in tareas]La lista conserva el orden de creación aunque las tareas terminen en otro orden.
TaskGroup frente a gather
asyncio.gather() sigue siendo útil cuando ya tienes awaitables y quieres un resultado ordenado. Con return_exceptions=True, puede devolver errores como valores. TaskGroup se centra en crear y poseer tareas dentro de un ámbito y cancela las hermanas cuando una falla.
No sustituyas mecánicamente todo gather(). Usa TaskGroup cuando las operaciones deban vivir y morir juntas; usa gather cuando su semántica exacta sea la deseada.
Añadir tareas dinámicamente
Mientras el contexto esté activo, una hija puede recibir el grupo y programar trabajo adicional.
async def descubrir(grupo: asyncio.TaskGroup, pagina: int) -> None:
await asyncio.sleep(0.1)
if pagina < 3:
grupo.create_task(descubrir(grupo, pagina + 1))
async def main():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(descubrir(grupo, 1))No se pueden añadir tareas después del cierre. Limita siempre la recursión, las colas y el fan-out.
Cancelación externa
Si se cancela la tarea que contiene el grupo, TaskGroup cancela sus hijas y espera su cierre. Las coroutines deben liberar recursos en finally.
async def consumidor():
recurso = await abrir_recurso()
try:
await procesar(recurso)
finally:
await recurso.cerrar()No captures CancelledError para continuar indefinidamente. Realiza cleanup y normalmente vuelve a lanzar la cancelación.
No ocultar CancelledError
TaskGroup y otras herramientas asyncio usan cancelación internamente. Consumir CancelledError puede romper el cierre estructurado.
async def incorrecto():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return # oculta la cancelaciónEl patrón seguro limpia y usa raise, salvo que consumir la cancelación sea una decisión deliberada y documentada.
Timeout para todo el grupo
Envuelve el ámbito con asyncio.timeout().
async def main():
try:
async with asyncio.timeout(2.0):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(operacion_a())
grupo.create_task(operacion_b())
except TimeoutError:
print("plazo excedido")Al expirar, se cancela la tarea contenedora; TaskGroup cancela y espera a las hijas.
Plazos individuales
Cuando cada operación necesita un límite propio, coloca el timeout dentro de la coroutine hija.
async def con_plazo(coro, segundos: float):
async with asyncio.timeout(segundos):
return await coroUn timeout individual genera TimeoutError en esa hija y normalmente cancela a las hermanas. Si es recuperable, trátalo dentro de la hija y devuelve un resultado explícito.
Fallas esperadas como datos
No todo error debe terminar el grupo. En procesamiento por lotes, un registro inválido puede convertirse en un resultado mientras los demás continúan.
from dataclasses import dataclass
@dataclass
class Resultado:
item_id: int
valor: str | None = None
error: str | None = None
async def procesar_seguro(item_id: int) -> Resultado:
try:
valor = await consultar_texto(item_id)
return Resultado(item_id=item_id, valor=valor)
except ErrorEsperado as exc:
return Resultado(item_id=item_id, error=str(exc))Reserva excepciones no tratadas para condiciones que invaliden toda la unidad concurrente.
Grupos anidados
TaskGroup puede representar suboperaciones.
async def procesar_cliente(cliente_id: int):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(cargar_perfil(cliente_id))
grupo.create_task(cargar_pedidos(cliente_id))
async def main(ids: list[int]):
async with asyncio.TaskGroup() as grupo:
for cliente_id in ids:
grupo.create_task(procesar_cliente(cliente_id))Cada nivel posee sus hijas. Las fallas pueden producir grupos de excepciones anidados que conservan la estructura del trabajo.
Limitar concurrencia
TaskGroup no limita la cantidad de tareas simultáneas. Crear cientos de miles puede consumir mucha memoria. Usa un semáforo, una cola de workers o lotes.
limite = asyncio.Semaphore(20)
async def limitado(item):
async with limite:
return await procesar(item)El grupo controla el ciclo de vida; el semáforo controla la capacidad.
Nombres de tareas y contexto
create_task() admite nombres y opciones de contexto según la versión de Python.
grupo.create_task(
consultar(42),
name="consulta-cliente-42",
)Los nombres ayudan en logs y debugging. Las context variables suelen copiarse a las tareas hijas.
Errores comunes
- Crear tareas relacionadas fuera del grupo: pueden escapar del ciclo de vida.
- Ocultar CancelledError: rompe la cancelación estructurada.
- Esperar una lista como retorno: conserva los Task para obtener resultados.
- Programar trabajo ilimitado: usa semáforos, colas o lotes.
- Tratar cada falla esperada como fatal: modela resultados recuperables como datos.
- Omitir cleanup: usa finally y context managers asíncronos.
Ejemplo completo: agregador de servicios
import asyncio
async def obtener_usuario(usuario_id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": usuario_id, "nombre": "Ana"}
async def obtener_pedidos(usuario_id: int) -> list[dict]:
await asyncio.sleep(0.2)
return [{"id": 1, "total": 99.0}]
async def construir_panel(usuario_id: int) -> dict:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as grupo:
usuario = grupo.create_task(obtener_usuario(usuario_id), name="usuario")
pedidos = grupo.create_task(obtener_pedidos(usuario_id), name="pedidos")
return {
"usuario": usuario.result(),
"pedidos": pedidos.result(),
}El panel se devuelve solo cuando ambas operaciones terminan. Una falla cancela a la hermana y evita devolver un resultado parcial accidental.
Conclusión
asyncio.TaskGroup organiza tareas relacionadas en un ámbito con inicio y fin claros. Espera a las hijas, coordina cancelación y agrupa fallas, haciendo el código asíncrono más predecible.
La documentación oficial de TaskGroup en asyncio explica excepciones y cancelación. Úsalo para operaciones que pertenecen juntas, limita la capacidad por separado y maneja cleanup y fallas esperadas de forma explícita.
