Entendendo Programação Assíncrona em Python
Antes de mergulharmos em httpx e aiohttp, precisamos compreender o fundamento que permite essas bibliotecas funcionarem: a programação assíncrona. Diferentemente da programação síncrona tradicional, onde uma operação bloqueia a execução até sua conclusão, a programação assíncrona permite que múltiplas operações rodem concorrentemente em uma única thread. Isso é especialmente útil para operações de I/O, como requisições HTTP, onde o tempo de espera pela resposta do servidor é considerável.
Python implementa isso através de coroutines (funções assíncronas) e um event loop. A palavra-chave async define uma função como coroutine, e await pausa a execução dessa função até que uma operação assíncrona seja concluída. Enquanto uma coroutine aguarda uma resposta, o event loop pode executar outras tarefas, maximizando a eficiência. Quando você precisa fazer 100 requisições HTTP, a abordagem assíncrona pode ser ordens de magnitude mais rápida que a síncrona, pois não espera cada resposta sequencialmente.
O Event Loop: Orquestrador de Tarefas
O event loop é o coração do sistema assíncrono. Ele gerencia todas as coroutines pendentes, alternando entre elas quando se deparam com operações de espera. Você acessa o event loop através de asyncio.get_event_loop() ou, em Python 3.10+, asyncio.run(), que gerencia automaticamente a criação e fechamento do loop.
import asyncio
async def tarefa_rapida():
print("Iniciando tarefa")
await asyncio.sleep(1) # Simula operação I/O
print("Tarefa concluída")
return "Resultado"
# Python 3.7+: forma recomendada
resultado = asyncio.run(tarefa_rapida())
print(resultado)
httpx: O Cliente HTTP Moderno
httpx é uma biblioteca HTTP de última geração que combina a simplicidade da API do requests com suporte nativo a requisições assíncronas. Ao contrário do requests tradicional, que é síncrono, httpx oferece uma interface consistente para tanto operações síncronas quanto assíncronas. A principal vantagem é que o mesmo código, com mínimas alterações, funciona nos dois modos.
Para instalar httpx, execute pip install httpx. A biblioteca é particularmente elegante porque sua API espelha requests, reduzindo a curva de aprendizado. Se você já conhece requests, adotar httpx é intuitivo. A classe AsyncClient é o ponto de entrada para operações assíncronas, oferecendo métodos como get(), post(), put(), delete(), etc., todos awaitable.
Realizando Requisições Simples com httpx
Começamos com um exemplo básico de uma requisição GET assíncrona:
import httpx
import asyncio
async def buscar_dados():
async with httpx.AsyncClient() as cliente:
resposta = await cliente.get("https://jsonplaceholder.typicode.com/posts/1")
print(f"Status: {resposta.status_code}")
print(f"Dados: {resposta.json()}")
asyncio.run(buscar_dados())
Observe que usamos async with para gerenciar o contexto do cliente. Isso garante que a conexão seja fechada corretamente após o uso. O método get() retorna um awaitable, por isso necessita do await. A resposta contém atributos como status_code, headers, content (bytes) e text (string), além do método json() para parsear JSON automaticamente.
Requisições Múltiplas: Paralelismo Real
A verdadeira força da programação assíncrona emerge quando você precisa fazer múltiplas requisições. Com httpx, é trivial:
import httpx
import asyncio
import time
async def buscar_multiplos():
ids = [1, 2, 3, 4, 5]
async with httpx.AsyncClient() as cliente:
# Cria tasks para cada requisição
tasks = [
cliente.get(f"https://jsonplaceholder.typicode.com/posts/{id}")
for id in ids
]
# Aguarda todas as tasks concorrentemente
respostas = await asyncio.gather(*tasks)
for i, resposta in enumerate(respostas):
print(f"Post {ids[i]}: {resposta.status_code}")
inicio = time.time()
asyncio.run(buscar_multiplos())
print(f"Tempo total: {time.time() - inicio:.2f}s")
Este padrão é essencial: asyncio.gather() recebe múltiplas coroutines e as executa concorrentemente, retornando uma lista com os resultados. Se cada requisição levasse 2 segundos, 5 requisições síncronas tomariam 10 segundos. Com esse código assíncrono, levam aproximadamente 2 segundos.
Tratamento de Erros e Timeouts
Requisições HTTP podem falhar. httpx oferece mecanismos robustos para tratamento:
import httpx
import asyncio
async def requisicao_segura():
async with httpx.AsyncClient(timeout=5.0) as cliente:
try:
resposta = await cliente.get("https://httpbin.org/delay/10")
resposta.raise_for_status() # Lança exceção se status >= 400
except httpx.TimeoutException:
print("Requisição expirou")
except httpx.HTTPStatusError as e:
print(f"Erro HTTP: {e.response.status_code}")
except httpx.RequestError as e:
print(f"Erro na requisição: {e}")
else:
print(f"Sucesso: {resposta.status_code}")
asyncio.run(requisicao_segura())
O parâmetro timeout define o tempo máximo de espera. Se excedido, TimeoutException é lançada. raise_for_status() converte códigos de erro HTTP em exceções. Sempre proteja requisições assíncronas com try-except adequado.
Requisições POST com Corpo e Headers Customizados
Frequentemente você precisa enviar dados:
import httpx
import asyncio
async def criar_recurso():
async with httpx.AsyncClient() as cliente:
payload = {
"title": "Novo Post",
"body": "Conteúdo do post",
"userId": 1
}
headers = {"User-Agent": "MeuCliente/1.0"}
resposta = await cliente.post(
"https://jsonplaceholder.typicode.com/posts",
json=payload, # Serializa automaticamente para JSON
headers=headers
)
print(f"Recurso criado: {resposta.json()}")
asyncio.run(criar_recurso())
Use json= para enviar dados que serão automaticamente serializados. Para dados de formulário, use data=. Headers customizados passam no parâmetro headers.
aiohttp: Alternativa Especializada para Websockets
aiohttp é outra excelente biblioteca para requisições HTTP assíncronas, frequentemente preferida em projetos que necessitam de websockets ou quando você quer uma dependência mais leve. Embora httpx seja mais completo e moderno, aiohttp ainda é bastante relevante na comunidade Python, especialmente em aplicações web real-time.
A instalação é pip install aiohttp. A diferença fundamental de aiohttp é que ela foi built from scratch para assincronismo, enquanto httpx começou oferecendo suporte assíncrono a uma API síncrona. Isso significa que aiohttp pode ter overhead ligeiramente menor em cenários de altíssima concorrência, embora na prática a diferença seja negligenciável para a maioria das aplicações.
Requisições Básicas com aiohttp
A sintaxe é similar, mas não idêntica:
import aiohttp
import asyncio
async def buscar_com_aiohttp():
async with aiohttp.ClientSession() as sessao:
async with sessao.get("https://jsonplaceholder.typicode.com/posts/1") as resposta:
print(f"Status: {resposta.status}")
dados = await resposta.json()
print(f"Dados: {dados}")
asyncio.run(buscar_com_aiohttp())
Observe o duplo async with: um para ClientSession (equivalente a AsyncClient no httpx) e outro para a resposta. Isso é uma peculiaridade de aiohttp — a resposta é um context manager que você deve usar explicitamente para ler o conteúdo.
Múltiplas Requisições com aiohttp
Padrão idêntico ao httpx:
import aiohttp
import asyncio
async def buscar_multiplos_aiohttp():
urls = [
f"https://jsonplaceholder.typicode.com/posts/{i}"
for i in range(1, 6)
]
async with aiohttp.ClientSession() as sessao:
tasks = [sessao.get(url) for url in urls]
respostas = await asyncio.gather(*tasks)
for resposta in respostas:
async with resposta:
print(f"Status: {resposta.status}")
asyncio.run(buscar_multiplos_aiohttp())
Websockets: Força Real de aiohttp
Onde aiohttp brilha é em websockets. httpx não possui suporte nativo a websockets, enquanto aiohttp oferece:
import aiohttp
import asyncio
async def usar_websocket():
async with aiohttp.ClientSession() as sessao:
async with sessao.ws_connect("wss://echo.websocket.org") as ws:
# Enviar mensagem
await ws.send_str("Olá, WebSocket!")
# Receber mensagem
msg = await ws.receive()
print(f"Resposta: {msg.data}")
# Este exemplo é ilustrativo; o servidor echo pode não estar disponível
# asyncio.run(usar_websocket())
Essa capacidade torna aiohttp indispensável para aplicações que necessitam comunicação bidirecional em tempo real.
Tratamento de Erros em aiohttp
import aiohttp
import asyncio
async def requisicao_segura_aiohttp():
timeout = aiohttp.ClientTimeout(total=5)
async with aiohttp.ClientSession(timeout=timeout) as sessao:
try:
async with sessao.get("https://httpbin.org/status/404") as resposta:
resposta.raise_for_status()
return await resposta.json()
except aiohttp.ClientConnectorError as e:
print(f"Erro de conexão: {e}")
except aiohttp.ClientPayloadError as e:
print(f"Erro ao ler resposta: {e}")
except asyncio.TimeoutError:
print("Requisição expirou")
asyncio.run(requisicao_segura_aiohttp())
httpx vs aiohttp: Qual Escolher?
Ambas as bibliotecas são maduras e confiáveis. A escolha depende do contexto do seu projeto. Use httpx se você precisa apenas de requisições HTTP simples e limpas — sua API é mais intuitiva e documentation melhor. Use aiohttp se você já está familiarizado com ela, precisa de websockets nativos, ou está em um projeto legado que já a utiliza.
Comparação Prática
# httpx: mais simples, mais moderno
import httpx
import asyncio
async def httpx_exemplo():
async with httpx.AsyncClient() as client:
r = await client.get("https://api.example.com/data")
return r.json()
# aiohttp: mais leve, melhor para websockets
import aiohttp
import asyncio
async def aiohttp_exemplo():
async with aiohttp.ClientSession() as session:
async with session.get("https://api.example.com/data") as r:
return await r.json()
# Ambas fazem a mesma coisa, mas com sintaxes diferentes
Para novos projetos recomendo httpx: melhor documentação, API mais consistente com requests que você provavelmente conhece, e suporte mais ativo. Para websockets, não há escolha — aiohttp é necessário.
Patterns Avançados
Pool de Conexões e Reutilização
Em aplicações de longa duração, reutilize o AsyncClient ou ClientSession:
import httpx
import asyncio
class APIClient:
def __init__(self):
self.client = None
async def connect(self):
self.client = httpx.AsyncClient()
async def disconnect(self):
await self.client.aclose()
async def buscar(self, url):
return await self.client.get(url)
# Uso
async def main():
api = APIClient()
await api.connect()
try:
resposta = await api.buscar("https://api.example.com/dados")
print(resposta.json())
finally:
await api.disconnect()
asyncio.run(main())
Isso evita criar e destruir clients repetidamente, reduzindo overhead.
Semaphore: Limitando Concorrência
Quando você tem muitas requisições, limitar concorrência evita sobrecarregar o servidor ou sua máquina:
import httpx
import asyncio
async def requisicoes_com_limite(urls):
semaforo = asyncio.Semaphore(5) # Máximo 5 requisições simultâneas
async def requisicao_limitada(url):
async with semaforo:
async with httpx.AsyncClient() as client:
return await client.get(url)
tasks = [requisicao_limitada(url) for url in urls]
return await asyncio.gather(*tasks)
# Uso com 100 URLs
urls = [f"https://api.example.com/item/{i}" for i in range(100)]
asyncio.run(requisicoes_com_limite(urls))
Retry com Backoff Exponencial
Requisições falham. Implementar retry inteligente é essencial:
import httpx
import asyncio
import time
async def requisicao_com_retry(url, max_tentativas=3):
for tentativa in range(max_tentativas):
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resposta = await client.get(url)
resposta.raise_for_status()
return resposta
except (httpx.RequestError, httpx.HTTPStatusError) as e:
if tentativa == max_tentativas - 1:
raise
espera = 2 ** tentativa # 1s, 2s, 4s
print(f"Tentativa {tentativa + 1} falhou. Aguardando {espera}s...")
await asyncio.sleep(espera)
# Uso
try:
resposta = asyncio.run(requisicao_com_retry("https://api.example.com/unstable"))
except Exception as e:
print(f"Falhou após retries: {e}")
Conclusão
Programação assíncrona com httpx e aiohttp transforma a eficiência de aplicações que lidam com requisições HTTP. O ponto fundamental é compreender que async/await permite que múltiplas operações de I/O rodem concorrentemente sem threads, usando um único event loop para orquestração. httpx é a escolha moderna e recomendada para a maioria dos casos por sua API limpa e intuitiva; aiohttp permanece relevante especialmente quando você necessita websockets. A diferença de desempenho entre ambas é mínima — escolha baseada em funcionalidades específicas e preferência da equipe.