Noções Básicas de Asyncio

9 exemplos para você começar com Asyncio - 6 básicos e 3 intermediários.

Pré-requisitos

• Python 3.14.0 com asyncio na biblioteca padrão.
• Use a sintaxe async/await - não os geradores legados @asyncio.coroutine.

Exemplos Básicos

1. Definição de Corrotina

Corrotinas são funções assíncronas pausadas em await.

import asyncio
 
async def hello() -> str:
    await asyncio.sleep(0.1)
    return "hello"
 
async def main() -> None:
    print(await hello())
 
asyncio.run(main())

• async def define uma função de corrotina.
• Chamá-la retorna um objeto corrotina - deve ser aguardada ou agendada.
• asyncio.run gerencia o ciclo de vida do loop para scripts.

Relacionado: Tarefas e Gather - corrotinas concorrentes

2. await Pausa Cooperativamente

Outras tarefas executam enquanto uma aguarda I/O.

import asyncio
 
async def tick(name: str) -> None:
    print(name, "start")
    await asyncio.sleep(0.1)
    print(name, "end")
 
async def main() -> None:
    await tick("a")
    await tick("b")
 
asyncio.run(main())

• await sequencial executa uma após a outra.
• sleep cede o controle para o loop de eventos.
• Sem paralelismo de CPU - apenas agendamento cooperativo.

Relacionado: Depurando Código Assíncrono - erros de bloqueio

3. gather para Concorrência

Execute múltiplas corrotinas concorrentemente.

import asyncio
 
async def work(n: int) -> int:
    await asyncio.sleep(0.1)
    return n
 
async def main() -> None:
    results = await asyncio.gather(work(1), work(2), work(3))
    print(results)
 
asyncio.run(main())

• gather agenda todos os awaitables no mesmo loop.
• Retorna uma lista na ordem de submissão quando todos são bem-sucedidos.
• Uma falha cancela as outras, a menos que return_exceptions=True.

Relacionado: Tarefas e Gather - objetos Task

4. Agendamento create_task de "Disparar e Esquecer"

Agende trabalho em segundo plano sem aguardar imediatamente.

import asyncio
 
async def background() -> None:
    await asyncio.sleep(0.1)
    print("done")
 
async def main() -> None:
    task = asyncio.create_task(background())
    print("scheduled")
    await task
 
asyncio.run(main())

• create_task envolve a corrotina em uma Tarefa (Task).
• Tarefas executam concorrentemente até serem aguardadas ou o loop parar.
• Armazene referências de tarefas para evitar GC em voo (caso raro).

Relacionado: Concorrência Estruturada - TaskGroup

5. asyncio.sleep vs time.sleep

Nunca bloqueie o loop com sleep síncrono.

import asyncio
 
async def good() -> None:
    await asyncio.sleep(0.05)
 
# time.sleep(0.05) em uma corrotina bloqueia TODAS as tarefas - evite

• asyncio.sleep é awaitable e não bloqueia outras tarefas.
• time.sleep congela todo o thread do loop de eventos.
• Mesma regra para drivers HTTP síncronos, DB síncronos em corrotinas.

Relacionado: Ponte Síncrono/Assíncrono - descarregar bloqueio

6. Timeout com wait_for

Limite o tempo de espera em corrotinas lentas.

import asyncio
 
async def slow() -> str:
    await asyncio.sleep(1)
    return "ok"
 
async def main() -> None:
    try:
        await asyncio.wait_for(slow(), timeout=0.1)
    except TimeoutError:
        print("timed out")
 
asyncio.run(main())

• Gera TimeoutError ao expirar.
• Cancela a tarefa subjacente no timeout.
• Prefira o gerenciador de contexto asyncio.timeout (3.11+) em código novo.

Relacionado: Concorrência Estruturada - padrões de timeout

Exemplos Intermediários

7. Padrão async with para Recursos

Gerenciadores de contexto assíncronos liberam recursos sem bloquear pares.

import asyncio
 
class AsyncResource:
    async def __aenter__(self):
        await asyncio.sleep(0)
        return self
 
    async def __aexit__(self, *args):
        await asyncio.sleep(0)
 
async def main() -> None:
    async with AsyncResource() as res:
        print("using", res)
 
asyncio.run(main())

• __aenter__/__aexit__ espelham o with síncrono.
• Use para conexões, locks, sessões.
• Veja a página dedicada para iteradores de streaming.

Relacionado: Gerenciadores e Iteradores de Contexto Assíncronos

8. Cheiro de Trabalho de CPU Vinculado ao Loop

Loops de CPU famintos deixam outras corrotinas sem recursos.

import asyncio
 
async def bad_cpu() -> int:
    total = 0
    for i in range(500_000):
        total += i
    return total
 
# Correção: await asyncio.to_thread(cpu_intensive_fn)

• Seções síncronas longas bloqueiam o "batimento cardíaco" e verificações de integridade.
• Descarregue para um pool de threads ou processos.
• Perfure antes de otimizar - loops pequenos são aceitáveis.

Relacionado: Ponte Síncrono/Assíncrono - to_thread

9. Entrada de Lifespan do FastAPI

Servidores mantêm um loop de longa duração - não asyncio.run repetido por solicitação.

# Padrão conceitual FastAPI 0.115+
# @asynccontextmanager async def lifespan(app):
#     yield
# app = FastAPI(lifespan=lifespan)

• O framework é dono do tempo de vida do loop.
• Abra httpx.AsyncClient compartilhado na inicialização do lifespan.
• Feche os recursos no desligamento.

Relacionado: HTTP Assíncrono com httpx/aiohttp - cliente compartilhado

Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.