O asyncio.Runner oferece uma forma estruturada de executar várias corrotinas de nível superior dentro do mesmo ciclo de eventos. Ele é útil em programas de linha de comando, ferramentas administrativas, testes, scripts de integração e aplicações que precisam chamar código assíncrono mais de uma vez sem criar e destruir toda a infraestrutura a cada chamada. Embora asyncio.run() continue sendo a escolha mais simples para uma única entrada assíncrona, o Runner melhora o controle quando o programa possui diversas fases.
Neste guia, você aprenderá como criar um Runner, reutilizar o loop, preservar variáveis de contexto, tratar sinais, habilitar debug, encerrar recursos e evitar erros comuns. O objetivo é mostrar não apenas a sintaxe, mas também um desenho seguro para aplicações reais.
Por que usar asyncio.Runner
asyncio.run() cria um novo event loop, executa uma corrotina e fecha o loop. Esse comportamento é excelente para programas com um único ponto de entrada. Porém, imagine uma ferramenta que precisa carregar configurações, sincronizar dados e depois gerar um relatório, sendo que cada etapa é assíncrona e deve compartilhar o mesmo ambiente. Chamar asyncio.run() repetidamente cria loops separados e impede o compartilhamento natural de tarefas, contexto e recursos.
O Runner resolve esse cenário ao encapsular o ciclo de vida do loop. Ele permite executar mais de uma corrotina sequencialmente, mantendo o mesmo loop e o mesmo contexto associado ao objeto. Isso reduz código manual e centraliza o encerramento.
Uso básico
import asyncio
async def carregar():
await asyncio.sleep(0.1)
return {"ambiente": "producao"}
async def sincronizar(config):
await asyncio.sleep(0.1)
return f"sincronizado em {config['ambiente']}"
with asyncio.Runner() as runner:
config = runner.run(carregar())
resultado = runner.run(sincronizar(config))
print(resultado)O gerenciador de contexto garante que o Runner seja fechado ao final. Cada chamada a run() recebe um awaitable e devolve seu resultado ou propaga a exceção. O loop permanece disponível entre as chamadas feitas dentro do bloco.
Runner não deve ser usado dentro de um loop ativo
Assim como asyncio.run(), o método Runner.run() não pode ser chamado quando já existe um event loop em execução na mesma thread. Isso acontece com frequência em notebooks, servidores assíncronos e frameworks que já controlam o loop. Nesses ambientes, use diretamente await ou integre a corrotina ao mecanismo do framework.
async def fluxo_existente():
resultado = await carregar()
return resultado
# Dentro de uma função async, use await.
# Não crie um Runner aqui.Essa regra evita loops aninhados e estados difíceis de prever. Antes de adicionar Runner a uma biblioteca, avalie quem é responsável pelo ciclo de eventos. Em geral, bibliotecas devem fornecer funções assíncronas; a aplicação executável decide como iniciar o loop.
Compartilhamento de ContextVar
Variáveis de contexto são úteis para IDs de correlação, informações de tenant e dados de observabilidade. O Runner mantém um contexto próprio que pode ser reutilizado entre chamadas. Também é possível passar explicitamente um contexto para run().
import asyncio
import contextvars
request_id = contextvars.ContextVar("request_id", default="sem-id")
async def registrar(etapa):
print(etapa, request_id.get())
ctx = contextvars.copy_context()
ctx.run(request_id.set, "job-847")
with asyncio.Runner() as runner:
runner.run(registrar("inicio"), context=ctx)
runner.run(registrar("fim"), context=ctx)Evite guardar segredos desnecessários em variáveis de contexto e limpe informações que não devem atravessar fases do programa. O contexto facilita propagação, mas não substitui uma política de segurança.
Modo de debug
O parâmetro debug=True habilita verificações adicionais do asyncio. Ele ajuda a identificar callbacks lentos, corrotinas nunca aguardadas e operações executadas na thread errada. Em desenvolvimento e testes, é uma ferramenta valiosa.
with asyncio.Runner(debug=True) as runner:
runner.run(carregar())O debug adiciona custo e pode gerar muito log. Em produção, habilite-o de maneira controlada, preferencialmente por configuração, e não exponha dados sensíveis nos registros.
Personalizando a criação do loop
O argumento loop_factory permite controlar como o event loop é criado. Isso pode ser necessário para instalar uma política específica, configurar instrumentação ou adaptar a execução a uma plataforma. A fábrica deve criar e registrar corretamente o loop.
import asyncio
def criar_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_debug(False)
return loop
with asyncio.Runner(loop_factory=criar_loop) as runner:
runner.run(carregar())Use essa opção apenas quando houver uma necessidade concreta. Uma fábrica incorreta pode causar recursos não fechados ou incompatibilidade com bibliotecas.
Tratamento de Ctrl+C
O Runner possui tratamento cuidadoso para KeyboardInterrupt. Quando o usuário pressiona Ctrl+C, a tarefa principal é cancelada para que a corrotina tenha oportunidade de executar blocos finally. Se o programa não terminar, uma nova interrupção pode encerrar de forma imediata.
async def servidor_temporario():
try:
while True:
await asyncio.sleep(1)
finally:
print("liberando recursos")
with asyncio.Runner() as runner:
runner.run(servidor_temporario())Não engula CancelledError sem uma razão clara. Faça a limpeza necessária e, em geral, propague o cancelamento. Bloquear o cancelamento pode impedir um encerramento previsível.
Encerrando tarefas e executores
Ao fechar, o Runner finaliza geradores assíncronos, encerra o executor padrão e fecha o loop. Isso reduz vazamentos de threads e descritores. Ainda assim, a aplicação deve controlar os próprios recursos: clientes HTTP, pools de banco, arquivos e filas precisam ser fechados explicitamente.
async def principal():
cliente = criar_cliente()
try:
await cliente.buscar_dados()
finally:
await cliente.aclose()Prefira gerenciadores de contexto assíncronos quando a biblioteca oferecer suporte. Eles tornam o ciclo de vida visível e testável.
Organizando múltiplas fases
Um bom uso do Runner é uma aplicação de terminal com fases separadas. Cada fase pode retornar dados para a próxima, enquanto o loop e o contexto permanecem estáveis.
async def validar():
return True
async def importar_dados():
return 120
async def emitir_relatorio(total):
print(f"{total} itens processados")
with asyncio.Runner() as runner:
if runner.run(validar()):
total = runner.run(importar_dados())
runner.run(emitir_relatorio(total))Não transforme cada função pequena em uma chamada separada ao Runner. Quando as operações fazem parte de um único fluxo assíncrono, é mais simples criar uma corrotina principal e usar await internamente. Use chamadas múltiplas quando houver fases realmente independentes controladas por código síncrono.
Concorrência estruturada
Dentro das corrotinas executadas, use asyncio.TaskGroup para coordenar tarefas relacionadas. Assim, falhas são propagadas de forma consistente e tarefas irmãs são canceladas quando necessário.
async def baixar(nome):
await asyncio.sleep(0.1)
return nome
async def lote():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(baixar("a"))
grupo.create_task(baixar("b"))Evite criar tarefas soltas sem guardar referência. Uma tarefa sem supervisão pode falhar silenciosamente ou continuar ativa no encerramento.
Timeouts e limites
Operações externas devem ter prazos. Use asyncio.timeout() para limitar blocos de trabalho e trate a expiração em um nível capaz de decidir se deve repetir, registrar ou abortar.
async def consultar():
async with asyncio.timeout(5):
return await chamada_remota()Combine timeout com limites de concorrência, tentativas com backoff e idempotência. O Runner organiza o loop, mas não resolve automaticamente sobrecarga ou dependências instáveis.
Testes
Em testes unitários assíncronos, prefira o suporte nativo do framework de testes. O Runner é mais útil para testar funções síncronas que coordenam etapas assíncronas. Garanta que cada teste feche o Runner e não compartilhe estado global entre casos.
Teste sucesso, exceção, cancelamento, timeout, Ctrl+C simulado quando relevante e liberação de recursos. Verifique também se nenhuma tarefa permanece pendente após o fechamento.
Boas práticas
- Use
asyncio.run()para uma única entrada simples. - Use Runner quando o código síncrono precisa executar várias fases assíncronas.
- Nunca chame Runner dentro de um loop já ativo.
- Feche clientes, pools e arquivos explicitamente.
- Propague cancelamentos após a limpeza.
- Use TaskGroup para tarefas relacionadas.
- Aplique timeouts e limites de concorrência.
- Ative debug em desenvolvimento.
- Evite loop_factory sem necessidade.
- Mantenha a responsabilidade pelo loop na camada de aplicação.
Continue estudando com os artigos da Academify sobre asyncio no Python, TaskGroup no Python, contextvars no Python e timeouts no asyncio.
Conclusão
asyncio.Runner é uma ferramenta de alto nível para controlar o ciclo de vida do asyncio em programas síncronos com várias etapas assíncronas. Ele reutiliza o event loop, organiza contexto, trata interrupções e executa o encerramento da infraestrutura. Seu melhor uso aparece em CLIs e ferramentas que possuem fases separadas. Para fluxos internos, continue usando uma corrotina principal, await, concorrência estruturada e gerenciamento explícito de recursos.
