
Passo 1 de 8
Delimitar a chamada que precisa sair do laço
Reconheça onde uma chamada síncrona interrompe o laço de eventos e decida se uma API assíncrona nativa ou uma ponte de execução é a escolha adequada.
Trilha de aprendizado · Nível 13 · Tutorial 10
Ao concluir, você será capaz de chamar funções síncronas a partir de corrotinas sem executá-las diretamente no laço de eventos, escolhendo o executor e o ciclo de vida adequados.
Delimitar a chamada que precisa sair do laço
Reconheça onde uma chamada síncrona interrompe o laço de eventos e decida se uma API assíncrona nativa ou uma ponte de execução é a escolha adequada. 2 min
Encaminhar entrada e saída com asyncio.to_thread
Execute uma função síncrona de entrada e saída fora da thread do laço e receba o resultado sem interromper outras corrotinas. 3 min
Integrar um executor explícito com run_in_executor
Submeta funções síncronas a um ThreadPoolExecutor explícito e aguarde seus resultados sem bloquear o laço de eventos. 3 min
Manter o encerramento bloqueante fora da corrotina
Organize um executor explícito no escopo síncrono que envolve asyncio.run, mantendo a espera pelos resultados cooperativa dentro do laço. 3 min
Usar processos sem mudar a forma de aguardar
Encaminhe processamento para processos separados sem alterar a fronteira assíncrona: o resultado continua sendo obtido com await. 3 min
Separar cancelamento da espera e término do trabalho
Observe por que um timeout pode encerrar a espera assíncrona enquanto a função síncrona continua em execução. 3 min
Preservar os recursos até o trabalhador terminar
Organize recursos usados por funções síncronas de modo que um timeout assíncrono não os feche enquanto o trabalhador ainda estiver em execução. 2 min
Aplicar a integração e verificar o ciclo completo
Reúna as pontes estudadas em um script local: entrada e saída em thread, processamento em processo e timeout que encerra a espera sem necessariamente encerrar o trabalhador. 3 min

Passo 1 de 8
Reconheça onde uma chamada síncrona interrompe o laço de eventos e decida se uma API assíncrona nativa ou uma ponte de execução é a escolha adequada.
Em uma corrotina, prefira uma API assíncrona nativa quando ela estiver disponível e atender ao problema. Ela já foi projetada para suspender a tarefa durante a espera, permitindo que o laço coordene outras tarefas.
Se a única API necessária for síncrona e puder demorar — por exemplo, uma leitura, uma consulta ou uma chamada a serviço externo — não a execute diretamente na thread do laço. A chamada deve atravessar uma ponte de execução para um trabalhador; a corrotina continua responsável por coordenar e aguardar o resultado.
Dica
Não basta uma função estar sendo chamada dentro de async def: se ela faz espera síncrona, ela ainda pode bloquear o laço. Localize a operação concreta que demora, não apenas a corrotina que a contém.
A diferença está em qual thread fica esperando pela operação síncrona.

À esquerda, a espera síncrona paralisa o laço. À direita, o laço mantém a coordenação enquanto um trabalhador realiza a espera.
Exemplo
Imagine uma corrotina que agenda uma tarefa para registrar um pulso periódico e, depois, precisa obter dados por uma biblioteca somente síncrona.
Os pulsos evidenciam que o laço continua progredindo. Eles não provam, por si só, que o programa ficou mais rápido: a operação ainda pode levar o mesmo tempo e ter outros custos.
Considere que cliente.buscar_registro pertence a uma biblioteca somente síncrona e pode esperar por rede:
async def mostrar_registro(cliente, codigo):
indicador = asyncio.create_task(mostrar_pulsos())
registro = cliente.buscar_registro(codigo)
await indicador
return registroQual é a chamada que precisa ser encaminhada para fora da thread do laço?
Uma biblioteca oferece buscar_async(codigo), uma operação assíncrona que fornece os mesmos dados de que você precisa. Também existe a versão síncrona buscar(codigo). Qual opção é preferível dentro da corrotina?

Passo 2 de 8
Execute uma função síncrona de entrada e saída fora da thread do laço e receba o resultado sem interromper outras corrotinas.
Quando não houver uma API assíncrona adequada e você precisar usar uma operação síncrona de entrada e saída, use asyncio.to_thread. Ela recebe a função e seus argumentos, devolvendo uma corrotina aguardável.
aguardavel = asyncio.to_thread(funcao_sincrona, arg1, opcao=True)
resultado = await aguardavelPasse funcao_sincrona, sem parênteses. Escrever funcao_sincrona() executaria a chamada imediatamente na thread do laço — justamente o que deve ser evitado.
A corrotina devolvida por to_thread começa a encaminhar o trabalho quando é aguardada ou quando é agendada como tarefa. O await entrega o valor retornado pela função ou propaga a exceção que ela produzir.
A ponte desloca a chamada síncrona; a coordenação das corrotinas permanece no laço de eventos.

A tarefa periódica pode progredir enquanto a operação síncrona espera fora da thread do laço.
Atenção
to_thread usa o executor padrão de threads; não há escolha de executor nessa chamada. Com o GIL habilitado, código Python limitado por processamento não ganha paralelismo de CPU apenas por usar threads. A ponte é especialmente útil para entrada e saída bloqueante — e também pode servir a extensões que liberam o GIL.
Crie um arquivo, por exemplo ponte_io.py, cole o código completo abaixo e execute python ponte_io.py. A função síncrona usa time.sleep apenas para simular uma espera de entrada e saída finita; uma aplicação real chamaria uma API síncrona necessária.
Observe que buscar_registro é passado sem parênteses e que tentativas=2 continua sendo um argumento nomeado da função síncrona.
import asyncio
import time
def buscar_registro(nome: str, *, tentativas: int) -> str:
print(f"[trabalhadora] iniciando busca de {nome!r}")
time.sleep(1.0) # Simula uma espera de entrada e saída.
print("[trabalhadora] busca concluída")
return f"{nome}: obtido após {tentativas} tentativas"
async def mostrar_progresso() -> None:
for numero in range(1, 5):
await asyncio.sleep(0.25)
print(f"[laço] pulso {numero}")
async def main() -> None:
ponte = asyncio.to_thread(
buscar_registro,
"relatorio-mensal",
tentativas=2,
)
tarefa_busca = asyncio.create_task(ponte)
await mostrar_progresso()
resultado = await tarefa_busca
print(f"[laço] resultado: {resultado}")
asyncio.run(main())Dica
Enquanto a trabalhadora está na espera simulada, devem aparecer pulsos do laço. A ordem exata de algumas linhas pode variar, mas os pulsos não devem ficar todos represados até o fim da busca. Se buscar_registro lançasse uma exceção, o await tarefa_busca a propagaria para main.
Complete a lacuna para encaminhar a função sem executá-la no laço:
resultado = await asyncio.to_thread(
_____,
"relatorio-mensal",
tentativas=2,
)Após executar o script, o que nos pulsos do terminal indica que a chamada síncrona não bloqueou o laço de eventos?
Escreva pelo menos 40 caracteres (0/40).

Passo 3 de 8
Submeta funções síncronas a um ThreadPoolExecutor explícito e aguarde seus resultados sem bloquear o laço de eventos.
Quando uma corrotina recebe um executor explícito, obtenha o laço que está em execução com asyncio.get_running_loop(). Em seguida, chame loop.run_in_executor(executor, funcao, *args).
Diferentemente de asyncio.to_thread, a chamada a run_in_executor já submete a função ao executor. Ela retorna um asyncio.Future, que pode ser aguardado com await.
A função síncrona sai da thread do laço; a corrotina continua coordenando a espera.

Com run_in_executor, a submissão ocorre na chamada e o resultado retorna por um futuro aguardável.
Dica
Neste step, considere que o ThreadPoolExecutor já foi criado por quem chamou a corrotina e foi passado como argumento. A criação e o encerramento corretos desse executor serão organizados no próximo step.
A função é passada sem parênteses; seus argumentos posicionais vêm depois dela.
import asyncio
from concurrent.futures import ThreadPoolExecutor
def ler_registro(codigo: int, tentativas: int) -> str:
# Representa uma biblioteca síncrona de entrada e saída.
return f"registro {codigo} lido em até {tentativas} tentativas"
async def consultar(executor: ThreadPoolExecutor) -> None:
loop = asyncio.get_running_loop()
futuro = loop.run_in_executor(executor, ler_registro, 42, 3)
print("trabalho já foi submetido")
resultado = await futuro
print(resultado)
get_running_loop() só é usado durante a execução assíncrona. run_in_executor(...) encaminha ler_registro e os argumentos 42 e 3 sem chamar ler_registro na thread do laço.
await futuro suspende apenas esta corrotina enquanto o resultado não chega. Se ler_registro falhar, a exceção é relançada no ponto do await.
Atenção
Evite loop.run_in_executor(executor, ler_registro(42, 3)). Nesse caso, ler_registro(42, 3) é executada imediatamente na thread do laço, e seu retorno — não a função — seria passado à API.
run_in_executor aceita argumentos posicionais após a função. Use partial para fixar argumentos nomeados antes da submissão.
import asyncio
from concurrent.futures import ThreadPoolExecutor
from functools import partial
def buscar_documento(codigo: int, *, idioma: str, incluir_anexos: bool) -> str:
return f"documento {codigo}: {idioma}, anexos={incluir_anexos}"
async def consultar(executor: ThreadPoolExecutor) -> None:
loop = asyncio.get_running_loop()
chamada = partial(
buscar_documento,
42,
idioma="pt-BR",
incluir_anexos=False,
)
futuro = loop.run_in_executor(executor, chamada)
print(await futuro)
Relacione cada expressão à sua característica principal.
Toque em um item e depois no par correspondente.
Dentro de uma corrotina, aguarde o retorno de run_in_executor com await futuro. Isso permite que o laço execute outras tarefas enquanto o trabalhador atua.
Não substitua essa espera por Future.result() de um futuro retornado por executor.submit: se o trabalho estiver pendente, essa chamada síncrona bloqueia a thread do laço.
Atenção
No asyncio.Future devolvido por run_in_executor, .result() não espera: se ele ainda não terminou, ela levanta asyncio.InvalidStateError. Mesmo depois de concluído, prefira await no fluxo assíncrono: ele obtém o retorno e propaga a exceção no ponto correto.
Após futuro = loop.run_in_executor(executor, ler_registro, 42, 3), qual linha obtém o resultado sem bloquear o laço enquanto o trabalho ainda está pendente?

Passo 4 de 8
Organize um executor explícito no escopo síncrono que envolve asyncio.run, mantendo a espera pelos resultados cooperativa dentro do laço.
Quando você cria um executor explícito para usar com run_in_executor, o código síncrono que chama asyncio.run é responsável por seu ciclo de vida.
Crie o executor, entre no with, execute asyncio.run(principal(executor)) e só então saia do contexto. Dentro da corrotina, use await para receber os resultados; fora dela, a saída do with faz o encerramento síncrono do executor.
A moldura externa representa o trecho síncrono; o laço de eventos vive apenas durante asyncio.run.

O encerramento do executor acontece depois que o laço de eventos já terminou.
Atenção
Evite colocar with ThreadPoolExecutor(...) dentro de async def. Ao sair desse contexto, o executor pode executar shutdown(wait=True). Essa espera é síncrona e pode paralisar o laço, impedindo outras tarefas assíncronas de progredirem.
Neste script, principal recebe o executor já criado. Ela aguarda cada futuro retornado por run_in_executor; falhas das funções síncronas são tratadas sem abandonar os demais resultados.
Salve como um arquivo .py e execute localmente.
import asyncio
from concurrent.futures import ThreadPoolExecutor
import time
def consultar_codigo(codigo: int) -> str:
time.sleep(0.2) # Simula uma operação síncrona de entrada e saída.
if codigo == 2:
raise ValueError("código indisponível")
return f"resultado do código {codigo}"
async def principal(executor: ThreadPoolExecutor) -> list[str]:
loop = asyncio.get_running_loop()
pendentes = [
loop.run_in_executor(executor, consultar_codigo, codigo)
for codigo in (1, 2, 3)
]
resultados: list[str] = []
for pendente in pendentes:
try:
resultados.append(await pendente)
except ValueError as erro:
resultados.append(f"falha tratada: {erro}")
return resultados
def main() -> None:
# O contexto é síncrono e envolve toda a vida do laço de eventos.
with ThreadPoolExecutor(max_workers=2) as executor:
resultados = asyncio.run(principal(executor))
# Aqui asyncio.run terminou e o executor já foi encerrado.
for resultado in resultados:
print(resultado)
if __name__ == "__main__":
main()
Dica
await pendente não bloqueia a thread do laço enquanto o trabalhador executa. Já a saída do with é uma operação síncrona de encerramento; por isso ela fica em main(), depois de asyncio.run(...).
Ordene as etapas de um programa que usa um ThreadPoolExecutor explícito com run_in_executor.
Por que o with ThreadPoolExecutor(...) deve envolver asyncio.run(...), em vez de ficar dentro da corrotina?
Escreva pelo menos 80 caracteres (0/80).

Passo 5 de 8
Encaminhe processamento para processos separados sem alterar a fronteira assíncrona: o resultado continua sendo obtido com await.
Quando uma função faz processamento intenso em código Python, um ProcessPoolExecutor pode executá-la em processos separados. Dentro da corrotina, a integração preserva a forma já usada com executor explícito:
resultado = await loop.run_in_executor(executor, funcao, argumento)
O executor muda de threads para processos; a forma de aguardar o resultado não muda. Processos podem ser uma escolha apropriada quando o GIL está habilitado, mas criar processos e transferir dados também tem custo. Portanto, isso não promete aceleração em todo caso.
Compare o que muda e o que permanece na integração.

O laço coordena e aguarda; o processamento ocorre nos processos trabalhadores.
Use um arquivo .py local e execute-o com Python. A função enviada ao processo deve estar no nível superior do módulo; argumentos e retorno precisam ser serializáveis. Neste exemplo, a lista de inteiros é enviada ao trabalhador, que devolve outro inteiro.
O executor é criado no escopo síncrono, com contexto spawn, e envolve asyncio.run(). A proteção if __name__ == "__main__" impede que trabalhadores recriem o programa ao importar o módulo principal.
Salve como processar_com_asyncio.py e execute: python processar_com_asyncio.py
import asyncio
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
def somar_quadrados(valores: list[int]) -> int:
# Esta função fica no nível superior para poder ser importada pelo processo.
return sum(valor * valor for valor in valores)
async def principal(executor: ProcessPoolExecutor) -> None:
loop = asyncio.get_running_loop()
valores = list(range(1, 100_001))
futuro = loop.run_in_executor(executor, somar_quadrados, valores)
resultado = await futuro
print(f"Soma dos quadrados: {resultado}")
def executar() -> None:
contexto = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(mp_context=contexto) as executor:
asyncio.run(principal(executor))
if __name__ == "__main__":
executar()Dica
run_in_executor(...) submete a função ao executor imediatamente. Já await futuro devolve o controle ao laço enquanto o processo trabalha e recebe o retorno ou a exceção quando a chamada termina.
Qual alternativa descreve corretamente uma integração de processamento com ProcessPoolExecutor?
Em duas ou três frases, explique por que o script usa ProcessPoolExecutor e por que ainda usa await para obter o resultado.
Escreva pelo menos 80 caracteres (0/80).

Passo 6 de 8
Observe por que um timeout pode encerrar a espera assíncrona enquanto a função síncrona continua em execução.
Quando uma corrotina aguarda uma função síncrona encaminhada a uma thread ou processo, há pelo menos dois eventos distintos:
Se a função já começou, cancelar a tarefa que a aguardava não oferece uma interrupção forçada da função em execução. O mesmo cuidado vale para threads e processos.
O prazo observado pela corrotina pode terminar antes do trabalho encaminhado.

O timeout corta a ligação de espera no laço; ele não prova que o trabalhador parou.
O script confirma que o trabalhador começou antes de aplicar o timeout. Os tempos servem apenas para tornar a observação provável; não use a ordem exata dos registros como contrato.
import asyncio
import threading
import time
def trabalho_lento(inicio: threading.Event) -> str:
print("trabalhador: iniciou")
inicio.set()
time.sleep(0.30) # simula uma operação síncrona finita
print("trabalhador: terminou")
return "resultado pronto"
async def main() -> None:
inicio = threading.Event()
tarefa = asyncio.create_task(asyncio.to_thread(trabalho_lento, inicio))
# Só iniciaremos o prazo depois de saber que a função começou.
await asyncio.to_thread(inicio.wait)
try:
async with asyncio.timeout(0.05):
resultado = await tarefa
print("corrotina: მიიღო", resultado)
except TimeoutError:
print("corrotina: timeout; a espera terminou")
print("tarefa cancelada?", tarefa.cancelled())
print("corrotina: fim de main")
asyncio.run(main())
print("programa: asyncio.run retornou")Atenção
É esperado observar tarefa cancelada? True e, ainda assim, ver trabalhador: terminou depois. O estado cancelado descreve a tarefa assíncrona que aguardava o resultado; não é um relatório confiável sobre o término da função síncrona já iniciada.
No exemplo, o executor padrão usado por asyncio.to_thread é gerenciado por asyncio.run. Ao encerrar, asyncio.run pode esperar o trabalho pendente desse executor. Por isso, a mensagem programa: asyncio.run retornou tende a aparecer somente após o término do trabalhador.
Com um executor explícito, seu encerramento também pode esperar trabalhadores em andamento. Chamar shutdown(wait=False) evita esperar naquela chamada, mas não interrompe funções que já estão executando nem garante que o processo saia imediatamente.
Se uma tarefa que aguarda asyncio.to_thread(...) fica cancelada após um timeout, então a função síncrona já iniciada certamente parou.
Considere este registro possível:
trabalhador: iniciou
corrotina: timeout; a espera terminou
tarefa cancelada? True
corrotina: fim de main
trabalhador: terminou
programa: asyncio.run retornouExplique por que esse registro não é contraditório e por que shutdown(wait=False) não seria prova de que o trabalhador parou.
Escreva pelo menos 80 caracteres (0/80).

Passo 7 de 8
Organize recursos usados por funções síncronas de modo que um timeout assíncrono não os feche enquanto o trabalhador ainda estiver em execução.
Depois de um timeout ou cancelamento, a corrotina parou de aguardar o resultado. Isso não prova que a função síncrona parou. Se ela ainda usa um arquivo, conexão ou outro recurso, fechá-lo no finally da corrotina pode interromper ou corromper o trabalho em andamento.
A regra prática é: quem executa o uso do recurso deve controlar sua vida útil até a saída real da função, inclusive quando ocorre uma exceção.

O fim da espera assíncrona acontece antes da limpeza feita pelo próprio trabalhador.
Atenção
Evite este desenho: a corrotina cria ou recebe um recurso, encaminha uma função que o usa e o fecha no finally logo após o timeout. O finally acompanha a espera da corrotina, não a execução efetiva do trabalhador.
Prefira passar uma descrição do recurso — por exemplo, um caminho de arquivo — e deixar a função síncrona abrir, usar e fechar o recurso com with. Assim, o fechamento ocorre quando a função realmente sai, seja por sucesso ou exceção.
Isso também é compatível com processos: caminhos, textos e números podem ser serializados; um arquivo já aberto não deve ser enviado ao ProcessPoolExecutor.
Exemplo
from pathlib import Path
import time
def ler_arquivo_lentamente(caminho: str) -> str:
# A função que usa o arquivo também o fecha.
with Path(caminho).open(encoding="utf-8") as arquivo:
time.sleep(2) # Simula uma operação bloqueante finita.
return arquivo.read()A corrotina passa apenas str(caminho) para asyncio.to_thread(...) ou para loop.run_in_executor(...). Se a espera assíncrona expirar após a função ter começado, o bloco with continua válido até ler_arquivo_lentamente retornar ou falhar.
Dica
Um timeout do asyncio não injeta uma interrupção dentro da função síncrona. Quando a operação oferece timeout próprio ou aceita um sinal de parada cooperativa, a própria função deve usá-los e sair; só então sua limpeza acontece. Não confunda abandonar a espera com encerrar o trabalho.
Um script cria um arquivo temporário e uma função síncrona precisa lê-lo lentamente em uma thread. A corrotina usa um timeout ao aguardar essa função. Qual organização preserva corretamente o recurso?
Resumo

Passo 8 de 8
Reúna as pontes estudadas em um script local: entrada e saída em thread, processamento em processo e timeout que encerra a espera sem necessariamente encerrar o trabalhador.
Em uma aplicação real, prefira uma API assíncrona nativa quando ela for adequada. Quando a dependência necessária for síncrona, mantenha a coordenação no laço e encaminhe o trabalho.
Neste exemplo, asyncio.to_thread recebe uma operação de entrada e saída simulada. Já run_in_executor envia uma função de processamento para um ProcessPoolExecutor. As duas operações são aguardadas com await, sem chamar funções bloqueantes diretamente na thread do laço.
A imagem resume quem executa cada parte e quando o encerramento pode esperar pelo trabalhador.

O timeout delimita a espera no laço; ele não é, por si só, um comando para interromper o processo trabalhador.
Salve o código como integracao.py e execute python integracao.py no seu computador. Ele usa apenas a biblioteca padrão e gera os próprios dados.
Os prints do processo usam flush=True para tornar mais visíveis os registros do trabalhador. A ordem exata entre mensagens pode variar.
Script completo com uma operação de entrada e saída simulada, uma operação de processamento e um timeout.
import asyncio
import multiprocessing
import time
from concurrent.futures import ProcessPoolExecutor
def ler_dado_local(nome: str, *, atraso: float) -> str:
"""Simula uma chamada síncrona limitada por entrada e saída."""
print(f"[thread] iniciando leitura de {nome}", flush=True)
time.sleep(atraso)
print(f"[thread] leitura concluída", flush=True)
return f"conteúdo de {nome}"
def somar_quadrados(valores: list[int], atraso: float) -> int:
"""Função de módulo, serializável para o processo trabalhador."""
print("[processo] processamento iniciado", flush=True)
time.sleep(atraso)
total = sum(valor * valor for valor in valores)
print("[processo] processamento terminado", flush=True)
return total
async def principal(executor: ProcessPoolExecutor) -> None:
loop = asyncio.get_running_loop()
leitura = asyncio.create_task(
asyncio.to_thread(ler_dado_local, "entrada.txt", atraso=0.2)
)
processamento = loop.run_in_executor(
executor,
somar_quadrados,
list(range(10)),
1.0,
)
try:
resultado_processamento = await asyncio.wait_for(
processamento,
timeout=0.3,
)
print(f"[laço] soma dos quadrados: {resultado_processamento}")
except TimeoutError:
print("[laço] timeout: a espera terminou; o processo pode continuar")
resultado_leitura = await leitura
print(f"[laço] resultado da leitura: {resultado_leitura}")
print("[laço] corrotina principal terminou")
if __name__ == "__main__":
contexto = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(mp_context=contexto) as executor:
asyncio.run(principal(executor))
print("[síncrono] executor encerrado; programa finalizado")Procure por estes fatos, não por tempos ou uma intercalação exata:
await.with ProcessPoolExecutor, que pode esperar o trabalhador terminar.Assim, o fim da espera não transfere para a corrotina a propriedade do trabalho nem autoriza encerrar recursos que o trabalhador ainda usa.
Após executar o arquivo, relate: por que cada ponte foi escolhida, onde o executor é encerrado e qual registro mostra que o trabalhador pode continuar depois do timeout.
Escreva pelo menos 180 caracteres (0/180).
Resumo
Use esta sequência ao integrar uma dependência síncrona a uma corrotina.
to_thread ou run_in_executor; não a execute antecipadamente no laço.await para obter retorno ou exceção sem bloquear a thread do laço.asyncio.run.Você precisa integrar uma biblioteca síncrona de entrada e saída e uma etapa de processamento Python demorada. Explique, de forma concisa, qual ponte usaria em cada caso e o que um timeout não garante.
Escreva pelo menos 120 caracteres (0/120).
Parabéns! Você concluiu: Integrar chamadas bloqueantes ao asyncio
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:.
+ 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