Criar várias tarefas com asyncio.create_task() é simples, mas coordenar o ciclo de vida delas pode ser difícil. Uma tarefa pode falhar enquanto outras continuam executando, referências podem ser perdidas e o encerramento pode acontecer antes do cleanup. asyncio.TaskGroup organiza tarefas assíncronas dentro de um bloco estruturado: ao sair do contexto, todas terminaram, foram canceladas ou tiveram seus erros reunidos.
Neste guia, você aprenderá a criar tarefas em grupo, coletar resultados, entender cancelamento, tratar ExceptionGroup, comparar TaskGroup com asyncio.gather(), aninhar grupos, aplicar timeouts e projetar serviços concorrentes previsíveis.
O problema das tarefas soltas
Uma tarefa criada com create_task() começa a executar independentemente. O chamador precisa guardar a referência, aguardar o resultado e lidar com falhas.
import asyncio
async def trabalho(nome: str) -> str:
await asyncio.sleep(0.2)
return nome.upper()
async def main():
tarefas = [
asyncio.create_task(trabalho("a")),
asyncio.create_task(trabalho("b")),
]
resultados = await asyncio.gather(*tarefas)
print(resultados)
asyncio.run(main())Esse padrão funciona, mas aumenta a chance de uma tarefa escapar do escopo lógico que a criou. TaskGroup torna o vínculo explícito.
Primeiro TaskGroup
import asyncio
async def trabalho(nome: str) -> str:
await asyncio.sleep(0.2)
return nome.upper()
async def main():
async with asyncio.TaskGroup() as grupo:
tarefa_a = grupo.create_task(trabalho("a"))
tarefa_b = grupo.create_task(trabalho("b"))
print(tarefa_a.result())
print(tarefa_b.result())
asyncio.run(main())O bloco só termina depois que todas as tarefas terminam. Após o async with, os resultados estão disponíveis e não existe trabalho pendente pertencente ao grupo.
Concorrência estruturada
Concorrência estruturada significa que tarefas filhas vivem dentro de um escopo claro. O código que cria as tarefas também é responsável por aguardar seu encerramento. Isso facilita raciocinar sobre recursos, cancelamento e erros.
O conceito complementa o guia de asyncio no Python. Corrotinas fornecem concorrência cooperativa; TaskGroup fornece organização para múltiplas corrotinas relacionadas.
Falha em uma tarefa
Quando uma tarefa do grupo falha com uma exceção diferente de cancelamento, TaskGroup normalmente cancela as outras tarefas ainda em execução. Depois que todas encerram, as falhas são levantadas como ExceptionGroup.
async def rapido():
await asyncio.sleep(0.1)
raise ValueError("dados inválidos")
async def lento():
try:
await asyncio.sleep(10)
finally:
print("cleanup do lento")
async def main():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(rapido())
grupo.create_task(lento())A tarefa lenta recebe cancelamento, executa o bloco finally e o grupo só então propaga o conjunto de erros.
Tratando ExceptionGroup com except*
async def main():
try:
async with asyncio.TaskGroup() as grupo:
grupo.create_task(rapido())
grupo.create_task(lento())
except* ValueError as grupo_erros:
for erro in grupo_erros.exceptions:
print("erro de validação:", erro)except* seleciona exceções compatíveis dentro de um grupo. Outros tipos não tratados continuam sendo propagados. Essa abordagem será aprofundada no conjunto sobre ExceptionGroup.
Coletando resultados
TaskGroup.create_task() devolve um objeto Task. Guarde as referências quando precisar dos resultados individuais.
async def consultar(id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": id}
async def carregar(ids: list[int]) -> list[dict]:
tarefas: list[asyncio.Task[dict]] = []
async with asyncio.TaskGroup() as grupo:
for id in ids:
tarefas.append(grupo.create_task(consultar(id)))
return [tarefa.result() for tarefa in tarefas]A ordem das tarefas na lista preserva a ordem de criação, mesmo que terminem em momentos diferentes.
TaskGroup versus gather
asyncio.gather() continua útil quando você já possui awaitables e deseja um único resultado ordenado. Com return_exceptions=True, ele pode coletar falhas como valores. TaskGroup, porém, oferece um modelo mais estruturado para criar tarefas dentro de um escopo e cancelar irmãs quando uma falha.
Não substitua mecanicamente todo gather(). Escolha TaskGroup para unidades de trabalho que devem viver e morrer juntas; use gather quando seu comportamento de agregação for exatamente o desejado.
Adicionando tarefas dinamicamente
Enquanto o grupo estiver ativo, uma corrotina pode receber o próprio grupo e adicionar tarefas.
async def descobrir(grupo: asyncio.TaskGroup, pagina: int) -> None:
await asyncio.sleep(0.1)
if pagina < 3:
grupo.create_task(descobrir(grupo, pagina + 1))
async def main():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(descobrir(grupo, 1))Novas tarefas só podem ser adicionadas enquanto o contexto está aberto. Evite crescimento sem limites: imponha profundidade, capacidade ou filas.
Cancelamento externo
Se a tarefa que contém o TaskGroup for cancelada, o grupo cancela suas tarefas filhas e aguarda o encerramento. Corrotinas devem respeitar CancelledError e liberar recursos em finally.
async def consumidor():
recurso = await abrir_recurso()
try:
await processar(recurso)
finally:
await recurso.fechar()Não capture CancelledError e continue indefinidamente. Se precisar executar cleanup, finalize e propague o cancelamento.
Não engolir CancelledError
TaskGroup e outras ferramentas de asyncio usam cancelamento internamente. Uma corrotina que captura BaseException ou CancelledError sem relançar pode quebrar o comportamento estruturado.
async def incorreto():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return # pode esconder o cancelamentoO padrão seguro é limpar e usar raise, salvo quando existe uma razão cuidadosamente documentada para consumir o cancelamento.
Timeout envolvendo o grupo
Use asyncio.timeout() para impor um prazo ao escopo inteiro.
async def main():
try:
async with asyncio.timeout(2.0):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(operacao_a())
grupo.create_task(operacao_b())
except TimeoutError:
print("prazo excedido")Ao expirar, a tarefa externa é cancelada; o TaskGroup cancela e aguarda as filhas. O próximo artigo do lote detalha o gerenciador de timeout.
Timeout individual por tarefa
Quando cada operação possui um prazo próprio, coloque o timeout dentro da corrotina filha.
async def com_prazo(coro, segundos: float):
async with asyncio.timeout(segundos):
return await coroUma expiração individual gera TimeoutError naquela tarefa e, por padrão, causa cancelamento das outras tarefas do grupo. Se a falha for recuperável, trate-a dentro da filha e retorne um resultado explícito.
Falhas esperadas como dados
Nem todo erro deve derrubar o grupo. Em um processamento em lote, talvez um item inválido deva produzir um resultado de falha enquanto os demais continuam.
from dataclasses import dataclass
@dataclass
class Resultado:
id: int
valor: str | None = None
erro: str | None = None
async def processar_seguro(id: int) -> Resultado:
try:
valor = await consultar_texto(id)
return Resultado(id=id, valor=valor)
except ErroEsperado as exc:
return Resultado(id=id, erro=str(exc))Reserve exceções não tratadas para condições que realmente invalidam a unidade concorrente.
Grupos aninhados
TaskGroups podem ser aninhados para representar suboperações.
async def processar_cliente(cliente_id: int):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(carregar_perfil(cliente_id))
grupo.create_task(carregar_pedidos(cliente_id))
async def main(ids: list[int]):
async with asyncio.TaskGroup() as grupo:
for id in ids:
grupo.create_task(processar_cliente(id))Cada nível controla suas próprias filhas. Falhas podem formar ExceptionGroups aninhados, preservando a estrutura do trabalho.
Limitar concorrência
TaskGroup não limita automaticamente a quantidade de tarefas simultâneas. Criar cem mil tarefas pode consumir muita memória. Use asyncio.Semaphore, uma fila com workers ou lotes.
limite = asyncio.Semaphore(20)
async def limitado(item):
async with limite:
return await processar(item)O grupo organiza o ciclo de vida; o semáforo controla capacidade.
Context variables e nomes de tarefas
create_task() aceita recursos como nome da tarefa e contexto, conforme a versão do Python. Nomes ajudam no diagnóstico.
grupo.create_task(
consultar(42),
name="consulta-cliente-42",
)Context variables normalmente são copiadas para a nova tarefa, permitindo request IDs e tracing por operação.
Erros comuns
- Criar tarefas fora do grupo sem necessidade: elas podem escapar do ciclo de vida.
- Engolir CancelledError: isso interfere no cancelamento estruturado.
- Esperar que TaskGroup retorne uma lista: guarde as Tasks para obter resultados.
- Criar tarefas sem limite: use semáforos, filas ou lotes.
- Tratar todas as falhas como exceções fatais: erros esperados podem virar dados.
- Ignorar cleanup: use finally e context managers assíncronos.
Exemplo completo: agregador de serviços
import asyncio
async def obter_usuario(id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": id, "nome": "Ana"}
async def obter_pedidos(id: int) -> list[dict]:
await asyncio.sleep(0.2)
return [{"id": 1, "total": 99.0}]
async def montar_painel(id: int) -> dict:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as grupo:
usuario = grupo.create_task(obter_usuario(id), name="usuario")
pedidos = grupo.create_task(obter_pedidos(id), name="pedidos")
return {
"usuario": usuario.result(),
"pedidos": pedidos.result(),
}O painel só é retornado depois que ambas as operações terminam. Uma falha cancela a operação irmã e evita retornar um estado parcial não planejado.
Conclusão
asyncio.TaskGroup organiza tarefas relacionadas em um escopo com início e fim claros. Ele aguarda todas as filhas, coordena cancelamento e reúne falhas, tornando o código assíncrono mais previsível.
A documentação oficial de TaskGroup no asyncio detalha propagação de exceções e cancelamento. Use grupos para operações que devem viver juntas, limite concorrência separadamente e trate cleanup e erros esperados de forma explícita.
