Criar tarefas com asyncio.create_task() é fácil; garantir que todas terminem, sejam canceladas e tenham exceções observadas é mais difícil. asyncio.TaskGroup, disponível desde Python 3.11, cria um limite claro: as tarefas pertencem ao bloco e são aguardadas ao sair dele.

Revise async e await em Python antes de aplicar concorrência a código de produção.

Executar tarefas relacionadas

import asyncio

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

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

    print(usuarios.result())
    print(pedidos.result())

asyncio.run(main())

O bloco só termina quando as duas tarefas acabam. Leia os resultados depois do async with, quando o ciclo de vida está completo.

Falhas e ExceptionGroup

Se uma tarefa gera uma exceção diferente de CancelledError, o grupo cancela as restantes. As falhas são propagadas em um ExceptionGroup:

try:
    async with asyncio.TaskGroup() as grupo:
        grupo.create_task(importar_clientes())
        grupo.create_task(importar_pedidos())
except* ValueError as grupo_erros:
    for erro in grupo_erros.exceptions:
        registrar(erro)

Use except* apenas quando puder tratar aquele conjunto de falhas. Não capture Exception indiscriminadamente nem engula cancelamentos. A documentação oficial de TaskGroup detalha a semântica de cancelamento.

TaskGroup ou gather

asyncio.gather() retorna resultados na ordem das corrotinas fornecidas e continua útil. TaskGroup oferece garantias mais fortes quando as tarefas formam uma unidade: se uma falha, as outras não ficam executando sem supervisão. Essa propriedade é chamada concorrência estruturada.

Não use concorrência para trabalho que precisa ser sequencial. Também não espere acelerar código pesado de CPU; nesse caso, estude multiprocessing e threads.

Cuidados práticos

  • Defina timeout no nível apropriado com asyncio.timeout().
  • Libere conexões e arquivos em finally ou context managers.
  • Não suprima CancelledError sem restaurar o cancelamento.
  • Dê nomes às tarefas quando isso ajudar observabilidade.
  • Limite concorrência ao chamar APIs e bancos.

TaskGroup torna explícito quais tarefas nascem e terminam juntas. Essa estrutura reduz tarefas órfãs e faz falhas concorrentes chegarem ao chamador de forma previsível.

Guardar referências e resultados

TaskGroup.create_task() devolve um objeto Task. Guardar essa referência é útil quando o chamador precisa associar cada resultado à operação que o produziu. Dentro do bloco, porém, a tarefa pode ainda estar em execução. A leitura de result() é segura depois da saída normal do contexto:

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

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

    return {
        cliente_id: tarefa.result()
        for cliente_id, tarefa in tarefas.items()
    }

O nome não muda a execução, mas ajuda em logs e ferramentas de diagnóstico. Se alguma consulta falhar, o fluxo não chega ao return: o grupo encerra as demais tarefas e propaga a falha. Portanto, não é necessário testar manualmente se cada tarefa terminou no caminho de sucesso.

Também é possível criar novas tarefas enquanto o grupo ainda está ativo, inclusive passando o próprio grupo a uma corrotina. Isso atende árvores de trabalho descobertas dinamicamente. Depois que a última tarefa termina e o contexto é encerrado, novas inclusões deixam de ser aceitas.

Cancelamento precisa cooperar

O cancelamento em asyncio não encerra uma corrotina à força. Ele injeta CancelledError no próximo ponto de suspensão, como um await. A corrotina deve liberar seus recursos e permitir que a exceção continue:

async def consumir_fila():
    conexao = await abrir_conexao()
    try:
        while True:
            item = await receber_item(conexao)
            await processar(item)
    finally:
        await conexao.close()

Evite capturar BaseException. Se for indispensável capturar CancelledError para registrar ou limpar estado, finalize o trabalho e use raise novamente. Componentes de concorrência estruturada, incluindo TaskGroup e asyncio.timeout(), usam cancelamento internamente; suprimi-lo pode atrasar o encerramento ou produzir comportamento confuso.

KeyboardInterrupt e SystemExit recebem tratamento especial: o grupo cancela e aguarda as outras tarefas, mas depois propaga a exceção original em vez de agrupá-la. Já falhas comuns são combinadas quando mais de uma tarefa consegue falhar durante o encerramento.

Tratar grupos de exceções sem esconder defeitos

except* divide um ExceptionGroup por tipo. Isso permite tratar erros recuperáveis e deixar os demais seguirem:

try:
    async with asyncio.TaskGroup() as grupo:
        for arquivo in arquivos:
            grupo.create_task(importar(arquivo))
except* ArquivoInvalido as erros:
    for erro in erros.exceptions:
        registrar_rejeicao(erro)
except* TimeoutError as erros:
    for erro in erros.exceptions:
        registrar_timeout(erro)

Tratar uma parte não apaga automaticamente as exceções de outros tipos. Elas continuam sendo propagadas, uma propriedade importante para não transformar erro de programação em sucesso aparente. Em muitos serviços, a melhor estratégia é registrar contexto perto da operação e deixar o grupo completo subir até uma camada que saiba decidir entre repetir, responder com erro ou encerrar o processo.

Não dependa da ordem das exceções para regras de negócio. A ordem de conclusão varia conforme rede, sistema operacional e carga. Se cada item puder falhar sem invalidar os demais, considere capturar o erro dentro da própria tarefa e retornar um resultado explícito, como Sucesso ou Falha. Nesse desenho, o grupo não vê a falha como exceção e não cancela os irmãos.

Aplicar timeout ao trabalho inteiro

Um timeout ao redor do grupo define um orçamento para a operação completa:

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

    return resumo.result(), alertas.result()

Se o prazo expirar, o contexto de timeout cancela a tarefa atual; o TaskGroup então encerra seus filhos antes de o TimeoutError chegar ao chamador. Um timeout por tarefa tem semântica diferente: uma chamada lenta pode falhar e, se a exceção escapar, provocar o cancelamento do grupo. Escolha o limite com base no compromisso do serviço, não apenas no tempo esperado de uma única dependência.

Para limitar quantidade simultânea, TaskGroup não substitui asyncio.Semaphore. Criar dez mil tarefas que aguardam um semáforo ainda cria dez mil objetos e mantém seus argumentos na memória. Para lotes muito grandes, combine um número fixo de workers com asyncio.Queue ou processe páginas menores.

Comparação cuidadosa com gather

gather() é conveniente quando você já possui uma lista fixa de awaitables e quer uma lista de resultados na mesma ordem. Com return_exceptions=True, ele pode representar falhas como valores, o que serve a operações independentes. Essa opção exige inspeção explícita de cada posição: ignorar um valor que é exceção cria falsos sucessos.

TaskGroup não devolve uma lista e não possui equivalente direto a return_exceptions=True. Em troca, relaciona criação, espera e cancelamento em uma região lexical. Essa é a escolha mais clara para subtarefas que só fazem sentido juntas, como montar partes obrigatórias de uma resposta.

Não envolva imediatamente toda chamada assíncrona em uma tarefa. Duas operações em que a segunda depende do resultado da primeira continuam sequenciais. A concorrência também pode pressionar pools de conexão, limites de API e memória. Meça latência e capacidade, mantenha limites explícitos e escreva testes que cubram falha e cancelamento, não somente o caminho feliz.

Testar o comportamento do grupo

Um teste útil provoca uma falha e confirma que a tarefa irmã executou sua limpeza. Use eventos para sincronizar o cenário, evitando testes baseados apenas em atrasos arbitrários. Também teste o contrato externo: qual exceção chega ao chamador, quais efeitos parciais são permitidos e se repetir a operação é seguro.

Para código compatível com versões anteriores ao Python 3.11, TaskGroup não está disponível na biblioteca padrão. Não esconda esse requisito: declare a versão mínima do projeto ou use uma solução compatível escolhida conscientemente. Em projetos modernos, o bloco estruturado costuma ser mais fácil de revisar do que uma coleção dispersa de create_task().