Passo 1 de 8
Definir o alcance do orçamento de tempo
Compare um prazo total compartilhado com limites renovados em cada operação.
Trilha de aprendizado · Nível 13 · Tutorial 9
Ao concluir, você será capaz de delimitar o tempo de espera de uma operação ou de um conjunto de operações assíncronas e tratar o esgotamento desse prazo sem ocultar cancelamentos externos.
Definir o alcance do orçamento de tempo
Compare um prazo total compartilhado com limites renovados em cada operação. 2 min
Delimitar um escopo com asyncio.timeout
Use um único orçamento de tempo para operações sequenciais e trate o prazo esgotado no ponto correto. 3 min
Limitar uma espera com asyncio.wait_for
Use wait_for para limitar a espera por um aguardável e identifique qual tarefa recebe o cancelamento quando o prazo se esgota. 2 min
Separar timeout, cancelamento externo e falha
Identifique a origem de cada desfecho e preserve cancelamentos externos e falhas reais ao tratar prazos locais. 3 min
Considerar o tempo de limpeza
Entenda por que o prazo configurado pode expirar antes de a operação e seu tratamento de timeout serem efetivamente concluídos. 2 min
Reconhecer os limites de um prazo cooperativo
Diagnostique por que um timeout pode ser observado depois do prazo nominal — ou nem produzir TimeoutError — quando o laço fica bloqueado ou o cancelamento é suprimido. 2 min
Aplicar um orçamento ao TaskGroup inteiro
Posicione um orçamento compartilhado ao redor de um TaskGroup para abranger a execução e o encerramento de todas as tarefas relacionadas. 3 min
Aplicar e verificar a política de tempo
Conclua uma aplicação local que combina limites individuais e um orçamento compartilhado, preservando cancelamentos externos e falhas reais. 3 min
Passo 1 de 8
Compare um prazo total compartilhado com limites renovados em cada operação.
Um tempo de espera define uma política: se o prazo for esgotado antes de a operação terminar, o programa deve interromper a espera e seguir o tratamento previsto. Ele não é uma pausa colocada antes de executar o trabalho.
A primeira decisão é o alcance desse prazo: ele vale para o conjunto inteiro de etapas ou recomeça a cada chamada?
Observe a diferença entre limitar a jornada inteira e dar um relógio novo a cada etapa.
No orçamento compartilhado, o tempo já gasto pela primeira etapa reduz o que resta para as seguintes.
Exemplo
Imagine três operações aguardadas em sequência:
Orçamento compartilhado de 3 s: as duas primeiras consomem 2,3 s; restam cerca de 0,7 s para montar a resposta. O requisito é: “a sequência toda deve caber em 3 s”.
Limite independente de 3 s por chamada: cada etapa pode usar até 3 s. Como os limites se renovam, a sequência pode chegar perto de 9 s sem violar nenhum limite individual. Isso não estabelece, por si só, um teto total de 3 s.
Dica
Pergunte: “Qual trabalho precisa terminar dentro deste prazo?” Se a resposta for o fluxo completo, escolha uma política de orçamento compartilhado. Se cada dependência tiver sua própria tolerância, limites independentes podem ser adequados.
Um endpoint executa, em sequência, validação, consulta e formatação. O requisito é responder em até 2 segundos no total. Qual política representa esse requisito?
Passo 2 de 8
Use um único orçamento de tempo para operações sequenciais e trate o prazo esgotado no ponto correto.
Use async with asyncio.timeout(segundos): para aplicar um orçamento de tempo a tudo que for aguardado dentro do bloco. As operações são sequenciais, mas compartilham o mesmo prazo: o tempo gasto na primeira reduz o disponível para a segunda.
O contexto não cria uma tarefa extra. Quando o prazo termina, ele solicita o cancelamento da tarefa corrente, que está executando o bloco.
As duas operações consomem o mesmo intervalo delimitado pelo contexto.
Um único prazo envolve as duas operações; não há reinício automático do contador entre elas.
Crie um arquivo, por exemplo escopo_timeout.py, e execute python escopo_timeout.py. O exemplo usa somente a biblioteca padrão e asyncio.sleep() para simular operações que suspendem cooperativamente.
import asyncio
async def operacao(nome: str, duracao: float) -> None:
print(f"início: {nome}")
await asyncio.sleep(duracao)
print(f"fim: {nome}")
async def main() -> None:
try:
async with asyncio.timeout(1.0):
await operacao("primeira", 0.4)
await operacao("segunda", 0.8)
except TimeoutError:
print("prazo do escopo esgotado")
else:
print("escopo concluído no prazo")
asyncio.run(main())Exemplo
A primeira operação consome cerca de 0,4 s. Restam cerca de 0,6 s para a segunda, que foi simulada para durar 0,8 s. Portanto, ela não chega ao seu fim e o programa imprime prazo do escopo esgotado.
Para observar o fluxo de sucesso, reduza a duração da segunda operação para 0.5. Nesse caso, as duas terminam e o bloco else é executado.
Ao atingir o prazo, o contexto solicita o cancelamento da tarefa corrente. Ao sair de async with, ele reconhece o cancelamento causado por seu próprio prazo e o apresenta como a exceção embutida TimeoutError.
Por isso, coloque except TimeoutError fora do async with. Dentro do bloco, o fluxo está sendo interrompido pelo cancelamento; é na saída do gerenciador que o resultado de prazo esgotado fica disponível como TimeoutError.
Dica
Neste caso, use except TimeoutError:. Em Python 3.12, TimeoutError é a exceção embutida apropriada para tratar o esgotamento do contexto.
Complete a linha que deve envolver o contexto para capturar corretamente o prazo esgotado:
async def main() -> None:
___:
async with asyncio.timeout(0.5):
await operacao("A", 0.2)
await operacao("B", 0.6)
except TimeoutError:
print("prazo esgotado")
Passo 3 de 8
Use wait_for para limitar a espera por um aguardável e identifique qual tarefa recebe o cancelamento quando o prazo se esgota.
Use await asyncio.wait_for(aguardavel, timeout) quando o prazo deve valer para a espera por um aguardável. O valor de timeout é dado em segundos.
Ao esgotar o prazo, wait_for solicita o cancelamento da operação aguardada — não da tarefa chamadora diretamente. Depois, aguarda o encerramento cooperativo dessa operação. Se ela encerra sem outra falha, a chamada resulta em TimeoutError.
Compare o alvo direto do cancelamento em cada política.
asyncio.timeout delimita a tarefa corrente; asyncio.wait_for delimita a espera por um aguardável.
Execute com Python 3.12 ou superior.
import asyncio
async def buscar_dados() -> str:
await asyncio.sleep(1)
return "dados recebidos"
async def main() -> None:
try:
resultado = await asyncio.wait_for(buscar_dados(), timeout=0.2)
print(resultado)
except TimeoutError:
print("A busca ultrapassou o prazo")
asyncio.run(main())buscar_dados() produz uma corrotina. Ao recebê-la, wait_for a agenda automaticamente como uma Task para poder acompanhá-la e cancelá-la se necessário.
Como a busca simula 1 segundo e o limite é 0,2, ela recebe o pedido de cancelamento. Após seu encerramento, wait_for lança a exceção embutida TimeoutError, capturada no try.
Dica
Se o aguardável já for uma Task, como uma criada com asyncio.create_task(...), wait_for a utiliza diretamente: não cria outra Task para a mesma operação.
Associe cada situação à descrição correta.
Toque em um item e depois no par correspondente.
Escolha wait_for quando a pergunta for: “quanto tempo posso esperar por este aguardável?”. Esse aguardável pode representar uma única operação ou trabalho composto.
Escolha um escopo asyncio.timeout quando várias esperas e trechos do bloco precisam compartilhar um único orçamento. Nos dois casos, o prazo delimita uma política de espera; o encerramento cooperativo é observado antes de a chamada retornar.
Você já possui uma Task que representa uma consulta e quer limitar apenas o tempo de espera por ela, sem aplicar automaticamente o mesmo prazo ao restante do código chamador. Qual escolha expressa essa fronteira?
Passo 4 de 8
Identifique a origem de cada desfecho e preserve cancelamentos externos e falhas reais ao tratar prazos locais.
Um prazo local é apenas uma das formas de uma espera terminar. Ao observar uma exceção, preserve sua origem:
Execute e analise esses cenários separadamente. Não tente decidir a origem com base em qual evento “chegou primeiro”.
A fronteira de tratamento deve reconhecer somente o prazo que ela própria definiu.
Timeout local, cancelamento externo e falha da operação exigem tratamentos diferentes.
Este script usa uma operação simulada. Execute um modo por vez, alterando a chamada em main().
import asyncio
async def operacao(modo: str) -> str:
await asyncio.sleep(0.1)
if modo == "falha":
raise ValueError("resposta inválida")
if modo == "timeout_proprio":
raise TimeoutError("prazo informado pela operação")
await asyncio.sleep(1)
return "resultado"
async def com_prazo_local(modo: str) -> None:
try:
async with asyncio.timeout(0.2):
resultado = await operacao(modo)
print(resultado)
except TimeoutError as erro:
print(f"timeout observado: {erro}")
async def main() -> None:
await com_prazo_local("falha")
asyncio.run(main())Atenção
O except TimeoutError acima observa tanto o prazo do contexto quanto um TimeoutError lançado por operacao. Portanto, o tipo isolado não prova a origem. Se essa diferença importar para sua regra de negócio, modele a falha da operação com uma exceção própria ou organize fronteiras em que a origem seja conhecida.
Use este padrão somente quando precisar registrar ou liberar recursos antes de propagar o cancelamento.
async def operacao_com_limpeza() -> None:
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
print("cancelamento externo observado; iniciando limpeza")
# Libere aqui os recursos que pertencem a esta operação.
raise # Nunca converta em sucesso nem em TimeoutError.Uma tarefa chama cancel() sobre a tarefa que executa com_prazo_local, antes de o prazo local expirar. Qual tratamento preserva corretamente o significado do evento?
Por que except BaseException seguido de uma mensagem como “timeout” é incorreto em uma função que usa asyncio.timeout? Explique o que deve acontecer com um CancelledError externo e com um ValueError produzido pela operação.
Escreva pelo menos 80 caracteres (0/80).
Passo 5 de 8
Entenda por que o prazo configurado pode expirar antes de a operação e seu tratamento de timeout serem efetivamente concluídos.
Quando um prazo se esgota, há três momentos distintos: (1) o prazo nominal vence; (2) o cancelamento é solicitado; (3) a operação termina seu encerramento. Se ela executa um finally, essa limpeza ocorre depois da solicitação e antes de o chamador observar TimeoutError.
Portanto, ultrapassar o valor configurado não prova que o timeout falhou: ele pode estar aguardando o encerramento cooperativo.
A limpeza vem após a expiração e antes da exceção chegar ao código que aguarda.
O intervalo de limpeza aumenta a duração observada, embora o prazo já tenha vencido.
Salve o script abaixo em um arquivo, por exemplo limpeza_timeout.py, e execute com Python 3.12 ou superior. As mensagens permitem observar a sequência; não use os valores de duração como uma medição exata.
import asyncio
async def operacao() -> None:
try:
print("operação: iniciada")
await asyncio.sleep(2)
finally:
print("operação: início da limpeza")
await asyncio.sleep(0.3)
print("operação: fim da limpeza")
async def main() -> None:
try:
await asyncio.wait_for(operacao(), timeout=0.1)
except TimeoutError:
print("chamador: timeout observado")
asyncio.run(main())Dica
A mensagem de início da limpeza aparece após o cancelamento solicitado pelo prazo. A mensagem chamador: timeout observado só aparece depois de operação: fim da limpeza, pois wait_for espera o encerramento da operação cancelada.
Com asyncio.timeout, o prazo cancela a tarefa corrente. Ao sair do async with, essa tarefa ainda precisa atravessar o finally da operação que estava aguardando. Só então o contexto converte seu próprio cancelamento em TimeoutError para o tratamento externo.
import asyncio
async def operacao() -> None:
try:
print("operação: iniciada")
await asyncio.sleep(2)
finally:
print("operação: início da limpeza")
await asyncio.sleep(0.3)
print("operação: fim da limpeza")
async def main() -> None:
try:
async with asyncio.timeout(0.1):
await operacao()
except TimeoutError:
print("chamador: timeout observado")
asyncio.run(main())Atenção
Em ambos os exemplos, o prazo não é uma garantia de retorno no instante exato configurado. Neste cenário cooperativo e sem outra falha, ele inicia o cancelamento no prazo e a limpeza prolonga a conclusão observada.
Em um cenário cooperativo com finally assíncrono, coloque os eventos na ordem correta.
Após executar um dos scripts, por que a mensagem de timeout aparece depois das mensagens de limpeza, mesmo com um prazo de 0,1 segundo?
Escreva pelo menos 60 caracteres (0/60).
Passo 6 de 8
Diagnostique por que um timeout pode ser observado depois do prazo nominal — ou nem produzir TimeoutError — quando o laço fica bloqueado ou o cancelamento é suprimido.
Um timeout configura um instante para solicitar cancelamento, mas o laço de eventos precisa voltar a executar para processar esse instante e entregar o cancelamento. Portanto, um prazo pequeno não é uma garantia de interrupção rígida naquele milissegundo.
Há três situações visualmente parecidas — o resultado aparece depois do prazo —, mas com causas diferentes: limpeza cooperativa, bloqueio do laço e supressão de CancelledError.
Limpeza adia a conclusão após o cancelamento; bloqueio atrasa a própria entrega do cancelamento; supressão faz a operação seguir ou retornar em vez de propagar o cancelamento.
time.sleep() não oferece um ponto de suspensão ao asyncio. Enquanto ela executa, o laço não processa o vencimento do timeout. No exemplo, o cancelamento só pode ser percebido depois que a chamada síncrona termina e a tarefa volta a fazer await.
Execute como um script local com Python 3.12 ou superior.
import asyncio
import time
async def main() -> None:
inicio = time.perf_counter()
try:
async with asyncio.timeout(0.05):
time.sleep(0.20) # Bloqueia o laço de eventos.
await asyncio.sleep(0) # O laço finalmente pode processar o prazo.
except TimeoutError:
decorrido = time.perf_counter() - inicio
print(f"Timeout observado após cerca de {decorrido:.2f} s")
asyncio.run(main())Atenção
O timeout continua sendo uma política de cancelamento cooperativo. Ele não consegue interromper uma chamada síncrona demorada nem processamento intenso sem pontos de suspensão. Para código bloqueante, será necessário escolher outra integração apropriada em um próximo tópico.
Se a operação capturar asyncio.CancelledError e retornar normalmente, o contexto de timeout pode terminar sem o TimeoutError esperado. Isso oculta o pedido de cancelamento e permite que um resultado improvisado pareça sucesso.
Quando precisar apenas observar ou limpar recursos, use finally ou capture CancelledError e relance-o com raise.
Aqui, o retorno dentro de except impede a propagação do cancelamento.
import asyncio
async def operacao() -> str:
try:
await asyncio.sleep(0.20)
return "resultado normal"
except asyncio.CancelledError:
print("Cancelamento recebido, mas suprimido indevidamente")
return "resultado improvisado" # Não faça isso para ocultar o cancelamento.
async def main() -> None:
try:
async with asyncio.timeout(0.05):
resultado = await operacao()
print(resultado)
except TimeoutError:
print("prazo esgotado")
asyncio.run(main())Dica
Se houver limpeza específica, mantenha-a em finally. Se capturar asyncio.CancelledError para registrar algo, termine o bloco com raise para preservar o cancelamento.
Relacione cada cenário ao motivo mais direto para o comportamento observado.
Toque em um item e depois no par correspondente.
asyncio.timeout(0.05) garante que qualquer código dentro do bloco pare exatamente após 0,05 segundo.
Passo 7 de 8
Posicione um orçamento compartilhado ao redor de um TaskGroup para abranger a execução e o encerramento de todas as tarefas relacionadas.
Posicione asyncio.timeout(...) por fora de asyncio.TaskGroup(). Assim, todas as tarefas criadas e a espera feita ao sair do grupo consomem o mesmo orçamento.
Quando o prazo vence, o contexto solicita o cancelamento da tarefa corrente. Ao sair do TaskGroup, esse cancelamento faz o grupo cancelar as tarefas ainda ativas e esperar que elas terminem a limpeza. Se esse encerramento cooperar e não houver outra falha, o TimeoutError é observado fora dos dois contextos.
O prazo externo cobre a execução e o encerramento estruturado.
A saída do TaskGroup precisa ficar dentro do asyncio.timeout: é nela que o grupo espera as tarefas restantes.
Crie um arquivo, por exemplo grupo_com_prazo.py, copie o código e execute python grupo_com_prazo.py. Os tempos são apenas uma simulação controlada: não dependa da ordem exata das mensagens entre tarefas.
import asyncio
async def trabalho(nome: str, duracao: float) -> None:
try:
print(f"{nome}: iniciou")
await asyncio.sleep(duracao)
print(f"{nome}: concluiu o trabalho")
finally:
print(f"{nome}: iniciou a limpeza")
await asyncio.sleep(0.10)
print(f"{nome}: terminou a limpeza")
async def main() -> None:
try:
async with asyncio.timeout(0.25):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(trabalho("rápida", 0.05))
grupo.create_task(trabalho("lenta", 1.00))
except TimeoutError:
print("orçamento do grupo esgotado")
asyncio.run(main())Exemplo
A tarefa lenta não termina seu trabalho de 1 segundo: ela recebe cancelamento quando o orçamento se esgota. Ainda assim, sua cláusula finally executa e aguarda a limpeza. Por isso, a mensagem orçamento do grupo esgotado pode aparecer depois de 0,25 segundo; o prazo nominal não elimina o tempo necessário para encerrar cooperativamente.
Complete a linha que deve vir imediatamente antes de async with asyncio.TaskGroup() as grupo:
____
No script, suponha que todas as limpezas terminem normalmente e nenhuma tarefa lance outra exceção. Qual resultado o except externo observa quando o prazo vence?
Atenção
Capture TimeoutError na fronteira da política de prazo, mas não use uma captura ampla para rotular falhas das tarefas como timeout. Um ExceptionGroup vindo do TaskGroup continua sendo evidência de falhas que exigem tratamento próprio.
Passo 8 de 8
Conclua uma aplicação local que combina limites individuais e um orçamento compartilhado, preservando cancelamentos externos e falhas reais.
Use asyncio.wait_for(...) quando quiser limitar a espera por um aguardável específico. Ao vencer o prazo, a operação aguardada recebe o pedido de cancelamento.
Use asyncio.timeout(...) ao redor de um bloco — inclusive de um TaskGroup — quando várias operações devem consumir o mesmo orçamento de tempo. Nesse caso, a tarefa corrente recebe o cancelamento; o encerramento estruturado alcança as tarefas do grupo.
Em ambos os casos, TimeoutError deve ser tratado na fronteira da política local. CancelledError externo e falhas como RuntimeError não devem ser rotulados como timeout.
Compare o alvo do cancelamento e o alcance do orçamento.
wait_for delimita uma espera; timeout delimita um escopo compartilhado.
Em um arquivo chamado politica_tempo.py, cole o código completo abaixo e execute python politica_tempo.py com Python 3.12 ou superior. Os atrasos são apenas simulações: observe as mensagens e as exceções, sem depender de valores exatos de duração.
import asyncio
async def operacao(
nome: str,
demora: float,
*,
falha: bool = False,
limpeza: float = 0,
) -> str:
print(f"{nome}: início")
try:
await asyncio.sleep(demora)
if falha:
raise RuntimeError(f"{nome}: falha própria da operação")
print(f"{nome}: resultado pronto")
return nome
finally:
print(f"{nome}: início da limpeza")
if limpeza:
await asyncio.sleep(limpeza)
print(f"{nome}: fim da limpeza")
async def limite_individual() -> None:
print("\n--- limite individual ---")
try:
resultado = await asyncio.wait_for(
operacao("consulta-lenta", 0.20, limpeza=0.05),
timeout=0.05,
)
print(f"resultado: {resultado}")
except TimeoutError:
print("chamador: prazo individual esgotado")
async def falha_da_operacao() -> None:
print("\n--- falha da operação ---")
try:
await asyncio.wait_for(
operacao("consulta-com-falha", 0.01, falha=True),
timeout=1,
)
except RuntimeError as erro:
print(f"chamador: falha preservada: {erro}")
async def cancelamento_externo() -> None:
print("\n--- cancelamento externo ---")
tarefa = asyncio.create_task(
asyncio.wait_for(operacao("consulta-cancelada", 1), timeout=2)
)
await asyncio.sleep(0.02)
tarefa.cancel()
try:
await tarefa
except asyncio.CancelledError:
print("chamador: cancelamento externo continuou se propagando")
async def orcamento_do_grupo() -> None:
print("\n--- orçamento compartilhado do grupo ---")
try:
async with asyncio.timeout(0.05):
async with asyncio.TaskGroup() as grupo:
grupo.create_task(operacao("rápida", 0.01))
grupo.create_task(
operacao("lenta-com-limpeza", 0.20, limpeza=0.05)
)
except TimeoutError:
print("chamador: orçamento do grupo esgotado")
async def main() -> None:
await limite_individual()
await falha_da_operacao()
await cancelamento_externo()
await orcamento_do_grupo()
asyncio.run(main())A execução deve mostrar, em cenários separados:
consulta-lenta inicia a limpeza antes de o chamador informar o timeout individual;RuntimeError da segunda operação chega ao chamador como falha, não como TimeoutError;tarefa.cancel() continua como CancelledError até o chamador que o observa;A ordem entre mensagens de tarefas concorrentes pode variar. O ponto importante é que a limpeza termina antes da observação final do timeout.
Após executar o arquivo, explique: qual política foi usada para cada cenário, qual exceção chegou ao chamador e como a limpeza afetou o encerramento.
Escreva pelo menos 180 caracteres (0/180).
Resumo
Antes de aplicar um prazo, defina a fronteira que realmente precisa ser limitada.
wait_for limita a espera por um aguardável; timeout dá um orçamento a um escopo.wait_for solicita cancelamento da operação aguardada; timeout cancela a tarefa corrente ao vencer.TimeoutError da política local, preserve falhas próprias e não suprima CancelledError externo.Parabéns! Você concluiu: Aplicar tempos de espera a operações assíncronas
Agora você consegue escolher entre um limite individual e um orçamento compartilhado, tratar o timeout na fronteira correta e preservar cancelamentos externos e falhas da operação.
Milhares de cursos online em vídeo, ebooks e áudiobooks.
Para testar seus conhecimentos no decorrer dos cursos online
Gerado diretamente na galeria de fotos do seu celular e enviado ao seu e-mail
Baixe nosso aplicativo pelo QR Code ou pelos links abaixo:.
Leve o conhecimento no seu bolso.
Baixe o aplicativo Cursa.
São centenas de Cursos gratuitos a disposição, com Certificado de conclusão gratuito que é salvo em sua galeria de imagens do celular.
Baixe o app para ter acesso ao Certificado de conclusão de Curso Gratuitamente.
+ de 10 milhões
de alunos
Certificado grátis e
válido em todo o Brasil
60 mil exercícios
gratuitos
4,8/5 classificação
nas lojas de apps
Cursos gratuitos em
vídeo, ebooks e audiobooks
