A programação assíncrona revolucionou a forma como desenvolvemos aplicações Python. Com a introdução do async e await no Python 3.5 (PEP 492), a linguagem ganhou uma ferramenta poderosa para escrever código concorrente de forma legível e eficiente. Neste guia completo, você vai dominar desde os fundamentos até técnicas avançadas de programação assíncrona.

Se você já trabalhou com chamadas de API, operações de I/O ou precisa processar múltiplas tarefas simultaneamente, dominar async/await é essencial para escrever código Python moderno e performático.

O Problema do Código Síncrono

No paradigma síncrono tradicional, cada operação bloqueia a execução até ser concluída. Isso é particularmente problemático para operações de I/O, como requisições HTTP, leitura de arquivos ou consultas a bancos de dados. Enquanto o programa aguarda a resposta de uma requisição de rede, a CPU fica ociosa, desperdiçando recursos preciosos.

import time
import requests
def buscar_dados(url):
print(f"Buscando {url}")
resposta = requests.get(url)
return resposta.status_code
inicio = time.time()
status1 = buscar_dados("https://api.github.com")
status2 = buscar_dados("https://api.github.com/users")
status3 = buscar_dados("https://api.github.com/repos")
fim = time.time()
print(f"Tempo total: {fim - inicio:.2f}s")

Neste exemplo, cada requisição espera a anterior terminar, resultando em um tempo total igual à soma de todas as latências. A programação assíncrona resolve esse problema permitindo que múltiplas operações sejam executadas concorrentemente.

O Que São Corrotinas?

Uma corrotina (coroutine) é uma função especial que pode pausar sua execução (await) e retomar posteriormente. Diferente de funções tradicionais, corrotinas permitem que o controle seja transferido voluntariamente em pontos específicos, possibilitando que outras tarefas sejam executadas durante a espera.

Para definir uma corrotina, basta usar async def no lugar de def:

async def minha_corrotina():
    print("Iniciando...")
    await alguma_operacao()
    print("Finalizado!")

A palavra-chave async transforma uma função comum em uma corrotina. Ao chamar uma corrotina, ela não executa imediatamente; em vez disso, retorna um objeto coroutine que precisa ser agendado no event loop. Mais detalhes sobre a definição de corrotinas podem ser encontrados na documentação oficial do CPython.

Async e Await na Prática

O grande poder do async/await está em executar operações de I/O sem bloquear a thread principal. Vamos reescrever o exemplo anterior usando a biblioteca aiohttp:

import asyncio
import aiohttp
import time
async def buscar_dados(sessao, url):
print(f"Buscando {url}")
async with sessao.get(url) as resposta:
return resposta.status
async def main():
async with aiohttp.ClientSession() as sessao:
tasks = [
buscar_dados(sessao, "https://api.github.com"),
buscar_dados(sessao, "https://api.github.com/users"),
buscar_dados(sessao, "https://api.github.com/repos"),
]
resultados = await asyncio.gather(*tasks)
print(f"Status: {resultados}")
inicio = time.time()
asyncio.run(main())
fim = time.time()
print(f"Tempo total: {fim - inicio:.2f}s")

Perceba como o tempo total agora é aproximadamente igual ao da operação mais lenta, não a soma de todas. Isso ocorre porque as requisições são executadas concorrentemente. A especificação completa do async/await está descrita na PEP 492 — Coroutines with async and await syntax.

O Módulo asyncio

O asyncio é o módulo padrão do Python para programação assíncrona. Ele fornece o event loop, que gerencia a execução de corrotinas, tasks e callbacks. O event loop é o coração da programação assíncrona, responsável por coordenar quando cada tarefa deve ser executada ou pausada.

import asyncio
async def saudacao(nome, delay):
await asyncio.sleep(delay)
print(f"Olá, {nome}!")
async def main():
Criando tarefas concorrentes
tarefa1 = asyncio.create_task(saudacao("Ana", 2))
tarefa2 = asyncio.create_task(saudacao("João", 1))
tarefa3 = asyncio.create_task(saudacao("Maria", 3))

# Aguardando todas as tarefas
await tarefa1
await tarefa2
await tarefa3
asyncio.run(main())

A função asyncio.run() foi introduzida no Python 3.7 e simplifica drasticamente a execução de corrotinas. Antes dela, era necessário gerenciar manualmente o event loop com loop.run_until_complete(). A documentação completa do módulo pode ser consultada em docs.python.org/3/library/asyncio.html.

Tasks e Futures

Uma Task é um wrapper ao redor de uma corrotina que a agenda para execução no event loop. Quando você chama asyncio.create_task(), a corrotina é automaticamente agendada para execução concorrente. Já um Future representa um resultado que estará disponível no futuro — é um conceito de nível mais baixo que raramente você precisará usar diretamente.

import asyncio
async def operacao_lenta(numero):
await asyncio.sleep(2)
return numero * 2
async def main():
Criando múltiplas tasks
tasks = [
    asyncio.create_task(operacao_lenta(1)),
    asyncio.create_task(operacao_lenta(2)),
    asyncio.create_task(operacao_lenta(3)),
]

# Aguardando conclusão
resultados = await asyncio.gather(*tasks)
print(f"Resultados: {resultados}")

# Verificando o status das tasks
for i, task in enumerate(tasks):
    print(f"Task {i}: concluída={task.done()}")
asyncio.run(main())

Tasks são fundamentais para executar múltiplas operações assíncronas em paralelo. A API completa de Tasks pode ser encontrada na documentação oficial de asyncio tasks.

Execução Concorrente com gather e as_completed

O asyncio.gather() é a ferramenta mais comum para executar múltiplas corrotinas concorrentemente. Ele retorna uma lista com os resultados na mesma ordem das corrotinas passadas. Já asyncio.as_completed() retorna resultados à medida que são concluídos, independentemente da ordem original.

import asyncio
async def fetchdata(id, delay):
await asyncio.sleep(delay)
return f"Dados do recurso {id_}"
async def main():
coros = [
fetch_data(1, 3),
fetch_data(2, 1),
fetch_data(3, 2),
]
# gather: resultados na ordem original
resultados = await asyncio.gather(*coros)
print(f"Gather: {resultados}")

# as_completed: resultados conforme finalizam
for coro in asyncio.as_completed(
    [fetch_data(4, 3), fetch_data(5, 1), fetch_data(6, 2)]
):
    resultado = await coro
    print(f"Finalizado: {resultado}")
asyncio.run(main())

Para operações mais complexas, asyncio.wait() permite controlar quando retornar (quando todas terminarem, quando a primeira terminar, ou quando a primeira falhar). Esses padrões são amplamente utilizados em aplicações reais e estão documentados na seção de execução concorrente de tarefas.

Timeouts e Tratamento de Erros

Em aplicações reais, é crucial lidar com operações que podem demorar mais do que o esperado. O asyncio.timeout() (disponível desde o Python 3.11) e asyncio.wait_for() permitem definir limites de tempo para suas operações assíncronas.

import asyncio
async def operacao_lenta():
await asyncio.sleep(10)
return "Concluído!"
async def main():
Usando timeout (Python 3.11+)
try:
    async with asyncio.timeout(3):
        resultado = await operacao_lenta()
        print(resultado)
except TimeoutError:
    print("Operação excedeu o tempo limite!")

# Usando wait_for
try:
    resultado = await asyncio.wait_for(
        operacao_lenta(), timeout=2
    )
    print(resultado)
except asyncio.TimeoutError:
    print("Timeout com wait_for!")
asyncio.run(main())

O tratamento de erros em código assíncrono segue o mesmo padrão try/except do Python síncrono, mas é importante notar que exceções em tasks precisam ser capturadas explicitamente. A PEP 525 — Asynchronous Generators define como generators assíncronos se comportam em relação a erros.

Requisições HTTP Assíncronas com aiohttp

Uma das aplicações mais comuns de async/await é fazer requisições HTTP concorrentes. A biblioteca aiohttp é a escolha padrão para HTTP assíncrono em Python, oferecendo suporte completo a clientes e servidores HTTP.

import asyncio
import aiohttp
async def consultar_api(sessao, url):
async with sessao.get(url) as resp:
dados = await resp.json()
return dados
async def main():
urls = [
"https://api.github.com",
"https://api.github.com/users/python",
"https://api.github.com/repos/python/cpython",
]
async with aiohttp.ClientSession() as sessao:
    tasks = [consultar_api(sessao, url) for url in urls]
    resultados = await asyncio.gather(*tasks)

    for url, dados in zip(urls, resultados):
        print(f"{url}: {len(dados)} campos")
asyncio.run(main())

O aiohttp também suporta sessões persistentes, conexões SSL, cookies e autenticação. A documentação completa está disponível em docs.aiohttp.org. Para quem prefere uma alternativa mais moderna, o httpx também oferece suporte a async com uma API similar à do requests.

Trabalhando com Arquivos Assíncronos

Operações de leitura e escrita de arquivos também se beneficiam da programação assíncrona, especialmente quando lidamos com múltiplos arquivos grandes. A biblioteca aiofiles fornece uma API assíncrona para operações de I/O em arquivos.

import asyncio
import aiofiles
async def processar_arquivo(caminho):
async with aiofiles.open(caminho, mode='r') as arquivo:
conteudo = await arquivo.read()
return f"{caminho}: {len(conteudo)} caracteres"
async def main():
arquivos = ["dados1.txt", "dados2.txt", "dados3.txt"]
tasks = [processar_arquivo(arq) for arq in arquivos]
resultados = await asyncio.gather(*tasks)
for resultado in resultados:
    print(resultado)
asyncio.run(main())

O aiofiles é particularmente útil para aplicações web que precisam servir arquivos grandes ou processar uploads sem bloquear o event loop. Mais informações podem ser encontradas no repositório oficial do aiofiles no GitHub.

Async Generators e Async Comprehensions

O Python também suporta generators assíncronos (PEP 525) e comprehensions assíncronas (PEP 530), que permitem produzir e consumir sequências de forma assíncrona. Esses recursos são úteis para processar streams de dados ou paginar resultados de APIs.

import asyncio
async def gerar_numeros():
for i in range(5):
await asyncio.sleep(0.5)
yield i
async def main():
Async comprehension
async for numero in gerar_numeros():
    print(f"Recebido: {numero}")

# List comprehension assíncrona
resultados = [x async for x in gerar_numeros()]
print(f"Resultados: {resultados}")
asyncio.run(main())

Os generators assíncronos implementam os protocolos __aiter__ e __anext__, e são fundamentais para bibliotecas de streaming e processamento de dados em tempo real. A especificação completa está disponível na PEP 530 — Asynchronous Comprehensions.

Boas Práticas com Async/Await

Escrever código assíncrono eficiente requer atenção a alguns princípios fundamentais:

1. Use asyncio.run() em vez de gerenciar o event loop manualmente

A função asyncio.run() gerencia a criação e o fechamento do event loop automaticamente, além de lidar com limpeza de recursos.

# Correto
asyncio.run(main())
Incorreto (versões antigas)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

2. Evite misturar código síncrono e assíncrono sem necessidade

Chamar funções síncronas dentro de corrotinas bloqueia o event loop. Se precisar executar código síncrono, use asyncio.to_thread() para executá-lo em uma thread separada.

import asyncio
import requests
async def buscar_dados_assincrono():
Isto BLOQUEIA o event loop!
resposta = requests.get("https://api.github.com")
return resposta.json()
async def buscar_dados_correto():
Isto NÃO bloqueia o event loop
loop = asyncio.get_running_loop()
resposta = await loop.run_in_executor(
    None, requests.get, "https://api.github.com"
)
return resposta.json()

3. Sempre use async context managers para recursos

Sessões HTTP, conexões de banco de dados e arquivos devem ser gerenciados com async with para garantir a liberação adequada de recursos.

async with aiohttp.ClientSession() as sessao:
    async with sessao.get(url) as resposta:
        return await resposta.json()

4. Cuidado com o excesso de tarefas concorrentes

Criar milhares de tasks simultaneamente pode sobrecarregar o event loop. Use asyncio.Semaphore para limitar o número de operações concorrentes.

import asyncio
async def processar_com_semaforo():
semaforo = asyncio.Semaphore(10)
async def tarefa_limitada(item):
    async with semaforo:
        await asyncio.sleep(1)
        return item * 2

tasks = [tarefa_limitada(i) for i in range(100)]
return await asyncio.gather(*tasks)

Para se aprofundar em padrões de concorrência, recomendamos a leitura do artigo sobre Async IO in Python do Real Python, que oferece uma excelente introdução ao tema com exemplos práticos.

Além disso, se você quer entender como decorators podem auxiliar na criação de código assíncrono mais limpo, confira nosso guia completo sobre Decorators em Python.

Async/Await em Frameworks Web

Os principais frameworks web Python modernos suportam nativamente async/await. O FastAPI foi construído desde o início com suporte a async, enquanto Django (desde a versão 3.1) e Flask (desde a versão 2.0) também oferecem suporte a views assíncronas. A combinação de async/await com frameworks web permite servir milhares de requisições simultaneamente com recursos mínimos.

Para dominar generators em Python, outra ferramenta essencial para processamento de dados em streaming, acesse nosso artigo sobre Generators em Python.

Conclusão

A programação assíncrona com async/await transformou a forma como escrevemos Python para operações de I/O. Com o módulo asyncio, bibliotecas como aiohttp e aiofiles, e as boas práticas apresentadas neste guia, você está preparado para escrever aplicações Python rápidas, eficientes e escaláveis.

Lembre-se: async/await não é sobre paralelismo real (múltiplas threads ou processos), mas sim sobre concorrência gerenciada por um event loop único. Para tarefas com uso intensivo de CPU, considere usar multiprocessing ou bibliotecas como concurrent.futures.

Para continuar seus estudos, explore a documentação oficial do asyncio, pratique com projetos reais e experimente diferentes bibliotecas assíncronas. O ecossistema async do Python cresce rapidamente, e dominar esses conceitos é um diferencial competitivo no mercado de desenvolvimento.