Timeout e cancelamento no asyncio impedem que uma operação lenta prenda a aplicação indefinidamente. Como o cancelamento é cooperativo, a tarefa recebe CancelledError no próximo ponto de espera e tem a oportunidade de liberar recursos. O código precisa respeitar esse sinal para não quebrar concorrência estruturada.

Este assunto parte do guia de async e await em Python. Para grupos de tarefas relacionadas, veja também asyncio.TaskGroup.

Limite um bloco com asyncio.timeout

Em Python 3.11 ou superior, asyncio.timeout() delimita várias operações com um único prazo:

import asyncio

async def carregar_pagina():
    try:
        async with asyncio.timeout(2.5):
            conexao = await abrir_conexao()
            return await conexao.receber()
    except TimeoutError:
        return None

Capture TimeoutError fora do bloco. Internamente, o context manager cancela a tarefa e transforma CancelledError ao sair. Para um prazo absoluto, calcule com o relógio monotônico do event loop e use asyncio.timeout_at().

wait_for limita um awaitable

resultado = await asyncio.wait_for(consultar_api(), timeout=2)

Quando o tempo acaba, wait_for cancela o awaitable e espera o processo de cancelamento. Por isso, o tempo real pode ultrapassar dois segundos se a limpeza demorar. asyncio.wait, por outro lado, devolve conjuntos de concluídas e pendentes sem cancelar automaticamente as pendentes.

Sempre libere recursos com finally

async def consumir(fila):
    recurso = await abrir_recurso()
    try:
        while True:
            item = await fila.get()
            await processar(item)
            fila.task_done()
    finally:
        await recurso.fechar()

Se você capturar asyncio.CancelledError para registrar ou limpar, use raise depois. Engolir a exceção pode fazer TaskGroup e asyncio.timeout() se comportarem de forma inesperada.

async def worker():
    try:
        await executar()
    except asyncio.CancelledError:
        logger.info("worker cancelado")
        raise

Cancele e aguarde a tarefa

Solicitar cancelamento não significa que a tarefa já terminou:

tarefa = asyncio.create_task(worker())

try:
    await usar_tarefa(tarefa)
finally:
    tarefa.cancel()
    try:
        await tarefa
    except asyncio.CancelledError:
        pass

Aguardar permite que o finally interno execute e evita tarefas órfãs. Em serviços, dê um prazo também ao encerramento e registre quando uma limpeza não termina.

Use shield com parcimônia

asyncio.shield() impede que o cancelamento do chamador seja propagado para uma tarefa interna, mas o chamador continua recebendo CancelledError. Isso pode fazer sentido para finalizar uma gravação crítica já iniciada. Guarde uma referência forte à tarefa e defina como seu resultado ou erro será observado.

Não use shield para mascarar uma arquitetura sem limites. Operações externas devem ter timeout próprio, e o desligamento deve ter um orçamento total.

Teste os caminhos lentos

Teste conclusão normal, timeout, cancelamento externo e erro durante a limpeza. Verifique que conexões, locks e itens de fila são liberados. Use esperas controladas em testes, não atrasos longos que deixam a suíte instável.

A documentação oficial de tarefas e timeouts do asyncio, consultada em 28 de julho de 2026, explica a transformação para TimeoutError, propagação de cancelamento e limites de shield.