Async em Python: asyncio, trio e trade-offs vs Node
⏱ 13 min·⭐ 55 XP
Pré-requisitos (0/1)0%
- ⬜🛡️ Pydantic v2 sério: modelos, validação e settings(Python para Engenheiros)
Recomendamos completar os pré-requisitos antes de seguir, mas nada te impede de continuar.
asyncio basics
| Tipo de trabalho | Ferramenta certa | Por quê |
|---|---|---|
| Espera de rede ou disco com biblioteca assíncrona | Laço de eventos | A espera cede o controle e outras tarefas avançam |
| Espera com biblioteca síncrona antiga | Empurrar para uma thread | A espera sai do laço; o resto continua atendendo |
| Cálculo pesado | Outro processo | O bloqueio do interpretador impede paralelismo real de bytecode entre threads |
| Muitas tarefas curtas relacionadas | Grupo de tarefas estruturado | Falha de uma cancela as irmãs, e o bloco não termina antes de todas |
| Uma tarefa disparada e esquecida | Evitar | Sobrevive ao escopo, e o erro dela some sem log |
Tipo de trabalhoEspera de rede ou disco com biblioteca assíncrona
Ferramenta certaLaço de eventos
Por quêA espera cede o controle e outras tarefas avançam
Tipo de trabalhoEspera com biblioteca síncrona antiga
Ferramenta certaEmpurrar para uma thread
Por quêA espera sai do laço; o resto continua atendendo
Tipo de trabalhoCálculo pesado
Ferramenta certaOutro processo
Por quêO bloqueio do interpretador impede paralelismo real de bytecode entre threads
Tipo de trabalhoMuitas tarefas curtas relacionadas
Ferramenta certaGrupo de tarefas estruturado
Por quêFalha de uma cancela as irmãs, e o bloco não termina antes de todas
Tipo de trabalhoUma tarefa disparada e esquecida
Ferramenta certaEvitar
Por quêSobrevive ao escopo, e o erro dela some sem log
import asyncio
import httpx
async def fetch_user(id: str) -> dict:
async with httpx.AsyncClient() as client:
r = await client.get(f"/users/{id}")
return r.json()
# Run (top-level)
async def main():
# Sequential — ruim
user_a = await fetch_user("a")
user_b = await fetch_user("b")
# Parallel — bom
user_a, user_b = await asyncio.gather(
fetch_user("a"),
fetch_user("b"),
)
asyncio.run(main())Evitar operação sync no event loop
# ❌ Trava o loop
async def bad():
import time
time.sleep(5) # loop morto por 5s
# ✅ Cede controle
async def good():
await asyncio.sleep(5)
# I/O legado sync:
async def with_legacy():
result = await asyncio.to_thread(legacy_blocking_fn, arg)
return result
# CPU-bound (processamento pesado):
from concurrent.futures import ProcessPoolExecutor
async def cpu_bound():
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as pool:
result = await loop.run_in_executor(pool, heavy_fn, arg)O laço
Laço de eventosuma thread só
Tarefa Aaguardando rede — cedeu o controle
Tarefa Bexecuta enquanto A espera
O que trava tudo
Chamada bloqueanteespera sem ceder — o laço para
Cálculo pesadonão há espera para intercalar
As saídas
Empurrar para uma threadpara I/O legado sem versão assíncrona
Outro processopara cálculo — o bloqueio do interpretador impede o resto
- → cede ao aguardar
- → trava
- → trava
- → a saída
- → a saída
- Conceito de arquitetura
- Compute
Retenha a assimetria: aguardar rede é grátis para os outros; qualquer outra espera é paga por todos. Encontrar a chamada bloqueante escondida é o diagnóstico mais comum em serviço assíncrono lento.
- 1 · Concorrência aqui é intercalar, não paralelizar. Uma thread executa tudo. Enquanto A aguarda a rede, B avança — e é só isso que acontece.
- 2 · Ceder o controle é responsabilidade de quem espera. Só quem aguarda de forma assíncrona devolve o laço. Uma espera bloqueante segura a única thread — e nenhuma outra tarefa avança.
- 3 · A palavra que marca a função não torna nada não bloqueante. Ela só permite aguardar dentro. Uma chamada síncrona ali dentro trava do mesmo jeito, e sem erro nenhum.
- 4 · I/O legado vai para uma thread. Biblioteca sem versão assíncrona pode ser empurrada para fora do laço. A espera acontece lá, e o laço segue atendendo.
- 5 · Cálculo pesado não se resolve com thread. O bloqueio do interpretador impede paralelismo real de bytecode. Para cálculo, a saída é outro processo.
- 6 · O sintoma é latência geral, não erro. Uma requisição lenta atrasa todas as outras. Como nada falha, a investigação procura no lugar errado — quase sempre no banco.
Structured concurrency (trio, anyio)
import anyio
# Task group — cancelamento estruturado
async def main():
async with anyio.create_task_group() as tg:
tg.start_soon(worker1)
tg.start_soon(worker2)
# Sai daqui APÓS ambos terminarem OU um falhar (cancelamento automático)
# anyio roda em cima de asyncio OU trio — backend-agnóstico
# Lib moderna (FastAPI usa anyio) prefere essa abordagem💡
trio popularizou structured concurrency (inspirou Swift async let). anyio unificou API pra escrever lib agnóstico do runtime. Em apps app, asyncio direto resolve; em libs, anyio.
Quiz rápido
Uma função `async` chama `time.sleep(5)`. Qual é o efeito?
Comparação Node ↔ Python async
- Node: event loop embutido, todo I/O é async por default. Promise <-> async/await.
- Python: loop explícito (asyncio.run). Biblioteca split entre sync e async versões (requests vs httpx.AsyncClient, psycopg vs asyncpg).
- Overhead: Python async é mais lento que Node em request/s raw. Mas para FastAPI + pydantic + DB, diferença some em workloads reais.
- GIL: I/O-bound libera GIL, paralelismo real. CPU-bound precisa de multiprocessing.
Perguntas frequentes
❓ Quando usar concorrência assíncrona em Python?
Quando o trabalho é de espera — rede, banco, arquivo — e há muitas operações simultâneas. Para trabalho de CPU, ela não ajuda: o laço fica bloqueado. Misturar chamada bloqueante dentro de função assíncrona é o defeito mais comum, e ele congela tudo.
❓ Como esperar várias operações assíncronas?
Com a função que reúne as tarefas quando você precisa de todas, e com a que entrega conforme completam quando quer processar o primeiro resultado. E com limite de concorrência, para não abrir mil conexões simultâneas contra um serviço que aguenta cem.
❓ Assíncrono em Python é como no Node?
O modelo é parecido — laço de eventos, um único fluxo de execução — e a diferença prática é que em Python a maior parte do ecossistema tem versão bloqueante e assíncrona. Escolher biblioteca bloqueante num serviço assíncrono anula o modelo inteiro.
Fixando
Quiz rápido
O que um task group de concorrência estruturada garante que disparar tarefas soltas não garante?
Quiz rápido
Para trabalho pesado de CPU em Python, por que `asyncio` não é a resposta?
Terminou de ler?
Marcar como concluído registra o XP, mantém sua sequência e coloca 3 cartas deste módulo na fila de revisão espaçada.
Próximos passos sugeridos
Temas deste módulo
Discussão
Carregando comentários…