PythonassíncioO módulo permite escrever código simultâneo usando a sintaxe async/await. Em 2026, o asyncio capacita FastAPI, Starlette e aiohttp. Este guia cobre tudo, desde loops de eventos até padrões do mundo real.
📋 Table of Contents
O que é assíncio?
asyncio usa um loop de eventos de thread único para gerenciar tarefas simultâneas — perfeito para E/S de rede e APIs onde você passa a maior parte do tempo esperando, não computando.
import asyncio
async def main():
print('Hello asyncio!')
await asyncio.sleep(1)
print('Done!')
asyncio.run(main())Sintaxe assíncrona/aguardada
Usarasync defpara definir uma corrotina. Usarawaitpara fazer uma pausa até que um resultado esteja pronto.
import asyncio
async def fetch_data(url: str) -> str:
await asyncio.sleep(0.5) # simulate network
return f'Data from {url}'
async def main():
result = await fetch_data('https://api.example.com')
print(result)
asyncio.run(main())Executando tarefas simultaneamente
asyncio.gather()executa várias corrotinas simultaneamente. Três tarefas de 1 segundo terminam em aproximadamente 1 segundo no total, não em 3.
import asyncio, time
async def task(name: str, delay: float):
print(f'Start {name}')
await asyncio.sleep(delay)
return name
async def main():
start = time.time()
results = await asyncio.gather(
task('A', 1.0),
task('B', 2.0),
task('C', 0.5),
)
print(f'Done in {time.time()-start:.2f}s: {results}')
# ~2.0s not 3.5s
asyncio.run(main())assíncrono com aiohttp
Busque 100 URLs simultaneamente, aproximadamente ao mesmo tempo que busca um.
import asyncio, aiohttp
async def fetch(session, url):
async with session.get(url) as resp:
return await resp.text()
async def main():
urls = ['https://httpbin.org/delay/1'] * 5
async with aiohttp.ClientSession() as session:
results = await asyncio.gather(*[fetch(session, u) for u in urls])
print(f'Fetched {len(results)} pages')
asyncio.run(main())asyncio.Queue para Produtor-Consumidor
import asyncio
async def producer(q: asyncio.Queue):
for i in range(5):
await q.put(f'item-{i}')
await asyncio.sleep(0.2)
await q.put(None) # sentinel
async def consumer(q: asyncio.Queue):
while True:
item = await q.get()
if item is None: break
print(f'Consumed {item}')
async def main():
q = asyncio.Queue(maxsize=3)
await asyncio.gather(producer(q), consumer(q))
asyncio.run(main())Tratamento de erros
import asyncio
async def risky(n):
if n == 2: raise ValueError(f'Task {n} failed!')
return f'Task {n} OK'
async def main():
results = await asyncio.gather(
*[risky(i) for i in range(4)],
return_exceptions=True
)
for r in results:
print('Error:', r if isinstance(r, Exception) else r)
asyncio.run(main())Melhores Práticas
- Nunca use
time.sleep()— usarawait asyncio.sleep() - Evite bloquear E/S dentro de corrotinas
- Usar
asyncio.run()como ponto de entrada - Usar
asyncio.TaskGroup(Python 3.11+) para simultaneidade estruturada - Depurar com
PYTHONASYNCIODEBUG=1
Conclusão
asyncio é a resposta do Python à simultaneidade no estilo Node.js. Mestreasyncio.gather(), adicione filas quando necessário e vá para TaskGroup para obter limites de erro mais limpos. Perfeito para APIs, scrapers e microsserviços.
🔗 Share this article
✍️ Leave a Comment