Operações assíncronas precisam de limites. Uma chamada de rede pode nunca responder, uma fila pode ficar vazia e um serviço lento pode consumir todo o tempo disponível de uma requisição. asyncio.timeout() cria um contexto com prazo: se o bloco não terminar a tempo, a tarefa é cancelada e o contexto transforma esse cancelamento em TimeoutError.
Neste guia, você aprenderá a usar timeouts relativos e absolutos, reagendar deadlines, verificar expiração, combinar com TaskGroup, comparar com asyncio.wait_for(), preservar cleanup e evitar erros comuns ao tratar cancelamento.
Primeiro timeout
import asyncio
async def operacao_lenta() -> str:
await asyncio.sleep(5)
return "ok"
async def main():
try:
async with asyncio.timeout(1.0):
resultado = await operacao_lenta()
print(resultado)
except TimeoutError:
print("tempo excedido")
asyncio.run(main())Quando um segundo passa, o contexto cancela a tarefa atual. Ao sair do bloco, o cancelamento interno é convertido em TimeoutError, que deve ser capturado fora do async with.
Por que capturar fora do contexto?
A transformação de CancelledError em TimeoutError acontece na saída do gerenciador. Dentro do bloco, o mecanismo opera por cancelamento.
try:
async with asyncio.timeout(1.0):
await operacao()
except TimeoutError:
...Esse padrão também deixa claro qual escopo possui o prazo.
Timeout não interrompe código bloqueante
Asyncio usa cancelamento cooperativo. O timeout só pode agir quando a coroutine entrega controle ao event loop. Código CPU-bound ou uma chamada bloqueante direta impede a reação.
async def incorreto():
time.sleep(10) # bloqueia o event loopPara funções bloqueantes, use asyncio.to_thread(), um executor ou um processo.
async with asyncio.timeout(2.0):
resultado = await asyncio.to_thread(funcao_bloqueante)O cancelamento deixa de aguardar a thread, mas não necessariamente interrompe a função que já está executando. Projete a operação bloqueante com seus próprios limites quando possível.
Timeout relativo e deadline absoluto
asyncio.timeout(segundos) define um limite relativo ao momento de entrada. asyncio.timeout_at(quando) recebe um instante do relógio monotônico do loop.
loop = asyncio.get_running_loop()
deadline = loop.time() + 3.0
async with asyncio.timeout_at(deadline):
await etapa_a()
await etapa_b()Deadlines absolutos são úteis quando várias camadas precisam compartilhar o mesmo orçamento sem reiniciar o cronômetro.
Propagando um orçamento de tempo
Imagine uma requisição com prazo total de cinco segundos. Cada função não deve ganhar cinco segundos novos; todas devem respeitar o mesmo deadline.
async def consultar_a(deadline: float):
async with asyncio.timeout_at(deadline):
return await chamada_a()
async def fluxo():
loop = asyncio.get_running_loop()
deadline = loop.time() + 5.0
a = await consultar_a(deadline)
b = await consultar_b(deadline)
return a, bO tempo gasto na primeira etapa reduz automaticamente o disponível para a segunda.
Reagendando o timeout
O objeto retornado pelo contexto possui reschedule(). Isso é útil quando o deadline só é conhecido depois de ler metadados ou negociar com outro sistema.
async def processar():
loop = asyncio.get_running_loop()
async with asyncio.timeout(None) as controle:
limite = await obter_limite()
controle.reschedule(loop.time() + limite)
return await executar_trabalho()None cria um contexto inicialmente sem prazo. O novo valor deve usar o relógio monotônico do loop.
Verificando se expirou
Após o contexto, expired() informa se o deadline foi atingido.
controle = None
try:
async with asyncio.timeout(1.0) as controle:
await trabalho()
except TimeoutError:
pass
if controle is not None and controle.expired():
print("o prazo expirou")Normalmente, capturar TimeoutError já é suficiente. expired() ajuda em métricas e diagnósticos.
Timeouts aninhados
Contextos podem ser aninhados com segurança. O prazo mais curto dispara primeiro.
async with asyncio.timeout(10.0):
await etapa_rapida()
try:
async with asyncio.timeout(1.0):
await chamada_opcional()
except TimeoutError:
usar_fallback()
await etapa_final()O timeout interno pode ser tratado localmente sem encerrar todo o fluxo, enquanto o timeout externo protege o orçamento total.
Combinando com TaskGroup
Um timeout ao redor de TaskGroup impõe prazo ao conjunto inteiro.
async def painel():
try:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as grupo:
usuario = grupo.create_task(obter_usuario())
pedidos = grupo.create_task(obter_pedidos())
except TimeoutError:
return {"erro": "serviços lentos"}
return {
"usuario": usuario.result(),
"pedidos": pedidos.result(),
}Ao expirar, a tarefa contenedora é cancelada; o grupo cancela as tarefas filhas e aguarda cleanup.
Timeout individual em um grupo
Coloque o timeout dentro da tarefa filha quando cada operação possui seu próprio limite.
async def consultar_com_limite(nome: str, segundos: float):
async with asyncio.timeout(segundos):
return await consultar(nome)Se essa função deixar TimeoutError escapar, TaskGroup trata como falha e cancela as irmãs. Para um timeout esperado, capture internamente e retorne um resultado explícito.
Comparação com asyncio.wait_for
asyncio.wait_for(awaitable, timeout) envolve um awaitable específico. asyncio.timeout() envolve um bloco inteiro, permitindo múltiplos awaits e lógica intermediária.
resultado = await asyncio.wait_for(operacao(), timeout=2.0)
async with asyncio.timeout(2.0):
cabecalho = await ler_cabecalho()
corpo = await ler_corpo(cabecalho)
await validar(corpo)O contexto costuma ser mais legível para fluxos compostos. wait_for() continua útil para uma única operação.
Cleanup e finally
A coroutine cancelada deve liberar recursos.
async def usar_conexao():
conexao = await abrir_conexao()
try:
return await conexao.receber()
finally:
await conexao.fechar()O timeout espera a propagação do cancelamento e a conclusão do cleanup. Um cleanup lento pode fazer a duração observada ultrapassar o prazo nominal. Mantenha encerramentos previsíveis e, quando necessário, dê limites próprios ao cleanup sem perder dados.
Não engolir CancelledError
Dentro da operação, o timeout é implementado com cancelamento. Capturar CancelledError e retornar normalmente impede que o contexto reconheça a expiração.
async def ruim():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return "ignorei"O padrão correto é executar cleanup e relançar.
except asyncio.CancelledError:
await limpar()
raiseShield e operações que não devem ser canceladas
asyncio.shield() pode proteger um awaitable interno do cancelamento da tarefa externa, mas exige cuidado.
tarefa = asyncio.create_task(gravar_auditoria())
try:
async with asyncio.timeout(1.0):
await asyncio.shield(tarefa)
except TimeoutError:
...
await tarefaO timeout deixa de aguardar, mas a tarefa protegida continua. Guarde uma referência forte e defina quem será responsável por esperá-la. Shield pode reintroduzir tarefas soltas se usado sem ownership claro.
Timeout não é retry
Um prazo encerra uma tentativa lenta; ele não decide se a operação deve ser repetida. Retentativas exigem política separada: quantidade máxima, backoff, jitter, idempotência e orçamento total.
async def tentar(deadline: float):
for tentativa in range(3):
try:
async with asyncio.timeout_at(deadline):
return await chamada()
except TimeoutError:
if tentativa == 2:
raise
await asyncio.sleep(0.1 * (2 ** tentativa))O deadline absoluto impede que retries ultrapassem o orçamento da requisição.
Testes de timeout
Evite testes que dependem de pausas longas. Use tempos curtos com margem razoável, corrotinas controladas por asyncio.Event e fakes que não completam até receber sinal.
async def nunca(evento: asyncio.Event):
await evento.wait()
async def teste():
evento = asyncio.Event()
try:
async with asyncio.timeout(0.05):
await nunca(evento)
except TimeoutError:
passNão verifique milissegundos exatos em ambientes de CI. Teste o comportamento: erro, cleanup e cancelamento das tarefas relacionadas.
Métricas e observabilidade
Diferencie timeout de conexão, leitura, operação, fila e orçamento total. Registre a duração, o deadline restante, o serviço chamado e o identificador da requisição. Não transforme todo timeout em erro genérico sem contexto.
Uma métrica alta de timeout pode indicar saturação, dependência lenta, limite agressivo ou event loop bloqueado.
Erros comuns
- Capturar TimeoutError dentro do contexto: a conversão ocorre ao sair.
- Usar time.sleep em coroutine: o loop fica bloqueado.
- Engolir CancelledError: o timeout pode não funcionar corretamente.
- Reiniciar o orçamento em cada camada: prefira deadline absoluto.
- Esperar que timeout mate uma thread: cancelamento assíncrono não interrompe código nativo arbitrário.
- Usar shield sem ownership: tarefas podem continuar sem supervisão.
Exemplo completo: cliente com orçamento total
import asyncio
async def consultar_servico(nome: str, deadline: float) -> dict:
async with asyncio.timeout_at(deadline):
await asyncio.sleep(0.2)
return {"servico": nome, "ok": True}
async def agregar() -> list[dict]:
loop = asyncio.get_running_loop()
deadline = loop.time() + 2.0
tarefas: list[asyncio.Task[dict]] = []
async with asyncio.timeout_at(deadline):
async with asyncio.TaskGroup() as grupo:
for nome in ["perfil", "pedidos", "saldo"]:
tarefas.append(
grupo.create_task(
consultar_servico(nome, deadline),
name=f"consulta-{nome}",
)
)
return [tarefa.result() for tarefa in tarefas]Todas as operações compartilham o mesmo deadline e pertencem ao mesmo grupo. O fluxo não cria um novo orçamento para cada serviço.
Conclusão
asyncio.timeout() transforma um prazo em um escopo claro. Ele usa cancelamento cooperativo, converte a expiração em TimeoutError e pode envolver várias operações relacionadas.
A documentação oficial de timeouts no asyncio detalha timeout(), timeout_at(), reagendamento e expiração. Use deadlines absolutos para orçamentos compartilhados, preserve cleanup e trate timeouts como parte de uma política maior de resiliência.
