Python

O que Todo Dev Deve Saber sobre httpx e aiohttp em Python: Requisições HTTP Assíncronas

16 min de leitura

O que Todo Dev Deve Saber sobre httpx e aiohttp em Python: Requisições HTTP Assíncronas

Entendendo Programação Assíncrona em Python Antes de mergulharmos em e , 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 (funções assíncronas) e um . A palavra-chave define uma função como coroutine, e 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

<h2>Entendendo Programação Assíncrona em Python</h2>

<p>Antes de mergulharmos em <code>httpx</code> e <code>aiohttp</code>, 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.</p>

<p>Python implementa isso através de <code>coroutines</code> (funções assíncronas) e um <code>event loop</code>. A palavra-chave <code>async</code> define uma função como coroutine, e <code>await</code> 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.</p>

<h3>O Event Loop: Orquestrador de Tarefas</h3>

<p>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 <code>asyncio.get_event_loop()</code> ou, em Python 3.10+, <code>asyncio.run()</code>, que gerencia automaticamente a criação e fechamento do loop.</p>

<pre><code class="language-python">import asyncio

async def tarefa_rapida():

print(&quot;Iniciando tarefa&quot;)

await asyncio.sleep(1) # Simula operação I/O

print(&quot;Tarefa concluída&quot;)

return &quot;Resultado&quot;

Python 3.7+: forma recomendada

resultado = asyncio.run(tarefa_rapida())

print(resultado)</code></pre>

<h2>httpx: O Cliente HTTP Moderno</h2>

<p><code>httpx</code> é uma biblioteca HTTP de última geração que combina a simplicidade da API do <code>requests</code> com suporte nativo a requisições assíncronas. Ao contrário do <code>requests</code> tradicional, que é síncrono, <code>httpx</code> 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.</p>

<p>Para instalar <code>httpx</code>, execute <code>pip install httpx</code>. A biblioteca é particularmente elegante porque sua API espelha <code>requests</code>, reduzindo a curva de aprendizado. Se você já conhece <code>requests</code>, adotar <code>httpx</code> é intuitivo. A classe <code>AsyncClient</code> é o ponto de entrada para operações assíncronas, oferecendo métodos como <code>get()</code>, <code>post()</code>, <code>put()</code>, <code>delete()</code>, etc., todos awaitable.</p>

<h3>Realizando Requisições Simples com httpx</h3>

<p>Começamos com um exemplo básico de uma requisição GET assíncrona:</p>

<pre><code class="language-python">import httpx

import asyncio

async def buscar_dados():

async with httpx.AsyncClient() as cliente:

resposta = await cliente.get(&quot;https://jsonplaceholder.typicode.com/posts/1&quot;)

print(f&quot;Status: {resposta.status_code}&quot;)

print(f&quot;Dados: {resposta.json()}&quot;)

asyncio.run(buscar_dados())</code></pre>

<p>Observe que usamos <code>async with</code> para gerenciar o contexto do cliente. Isso garante que a conexão seja fechada corretamente após o uso. O método <code>get()</code> retorna um <code>awaitable</code>, por isso necessita do <code>await</code>. A resposta contém atributos como <code>status_code</code>, <code>headers</code>, <code>content</code> (bytes) e <code>text</code> (string), além do método <code>json()</code> para parsear JSON automaticamente.</p>

<h3>Requisições Múltiplas: Paralelismo Real</h3>

<p>A verdadeira força da programação assíncrona emerge quando você precisa fazer múltiplas requisições. Com <code>httpx</code>, é trivial:</p>

<pre><code class="language-python">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&quot;https://jsonplaceholder.typicode.com/posts/{id}&quot;)

for id in ids

]

Aguarda todas as tasks concorrentemente

respostas = await asyncio.gather(*tasks)

for i, resposta in enumerate(respostas):

print(f&quot;Post {ids[i]}: {resposta.status_code}&quot;)

inicio = time.time()

asyncio.run(buscar_multiplos())

print(f&quot;Tempo total: {time.time() - inicio:.2f}s&quot;)</code></pre>

<p>Este padrão é essencial: <code>asyncio.gather()</code> 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.</p>

<h3>Tratamento de Erros e Timeouts</h3>

<p>Requisições HTTP podem falhar. <code>httpx</code> oferece mecanismos robustos para tratamento:</p>

<pre><code class="language-python">import httpx

import asyncio

async def requisicao_segura():

async with httpx.AsyncClient(timeout=5.0) as cliente:

try:

resposta = await cliente.get(&quot;https://httpbin.org/delay/10&quot;)

resposta.raise_for_status() # Lança exceção se status &gt;= 400

except httpx.TimeoutException:

print(&quot;Requisição expirou&quot;)

except httpx.HTTPStatusError as e:

print(f&quot;Erro HTTP: {e.response.status_code}&quot;)

except httpx.RequestError as e:

print(f&quot;Erro na requisição: {e}&quot;)

else:

print(f&quot;Sucesso: {resposta.status_code}&quot;)

asyncio.run(requisicao_segura())</code></pre>

<p>O parâmetro <code>timeout</code> define o tempo máximo de espera. Se excedido, <code>TimeoutException</code> é lançada. <code>raise_for_status()</code> converte códigos de erro HTTP em exceções. Sempre proteja requisições assíncronas com try-except adequado.</p>

<h3>Requisições POST com Corpo e Headers Customizados</h3>

<p>Frequentemente você precisa enviar dados:</p>

<pre><code class="language-python">import httpx

import asyncio

async def criar_recurso():

async with httpx.AsyncClient() as cliente:

payload = {

&quot;title&quot;: &quot;Novo Post&quot;,

&quot;body&quot;: &quot;Conteúdo do post&quot;,

&quot;userId&quot;: 1

}

headers = {&quot;User-Agent&quot;: &quot;MeuCliente/1.0&quot;}

resposta = await cliente.post(

&quot;https://jsonplaceholder.typicode.com/posts&quot;,

json=payload, # Serializa automaticamente para JSON

headers=headers

)

print(f&quot;Recurso criado: {resposta.json()}&quot;)

asyncio.run(criar_recurso())</code></pre>

<p>Use <code>json=</code> para enviar dados que serão automaticamente serializados. Para dados de formulário, use <code>data=</code>. Headers customizados passam no parâmetro <code>headers</code>.</p>

<h2>aiohttp: Alternativa Especializada para Websockets</h2>

<p><code>aiohttp</code> é 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 <code>httpx</code> seja mais completo e moderno, <code>aiohttp</code> ainda é bastante relevante na comunidade Python, especialmente em aplicações web real-time.</p>

<p>A instalação é <code>pip install aiohttp</code>. A diferença fundamental de <code>aiohttp</code> é que ela foi built from scratch para assincronismo, enquanto <code>httpx</code> começou oferecendo suporte assíncrono a uma API síncrona. Isso significa que <code>aiohttp</code> 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.</p>

<h3>Requisições Básicas com aiohttp</h3>

<p>A sintaxe é similar, mas não idêntica:</p>

<pre><code class="language-python">import aiohttp

import asyncio

async def buscar_com_aiohttp():

async with aiohttp.ClientSession() as sessao:

async with sessao.get(&quot;https://jsonplaceholder.typicode.com/posts/1&quot;) as resposta:

print(f&quot;Status: {resposta.status}&quot;)

dados = await resposta.json()

print(f&quot;Dados: {dados}&quot;)

asyncio.run(buscar_com_aiohttp())</code></pre>

<p>Observe o duplo <code>async with</code>: um para <code>ClientSession</code> (equivalente a <code>AsyncClient</code> no <code>httpx</code>) e outro para a resposta. Isso é uma peculiaridade de <code>aiohttp</code> — a resposta é um context manager que você deve usar explicitamente para ler o conteúdo.</p>

<h3>Múltiplas Requisições com aiohttp</h3>

<p>Padrão idêntico ao <code>httpx</code>:</p>

<pre><code class="language-python">import aiohttp

import asyncio

async def buscar_multiplos_aiohttp():

urls = [

f&quot;https://jsonplaceholder.typicode.com/posts/{i}&quot;

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&quot;Status: {resposta.status}&quot;)

asyncio.run(buscar_multiplos_aiohttp())</code></pre>

<h3>Websockets: Força Real de aiohttp</h3>

<p>Onde <code>aiohttp</code> brilha é em websockets. <code>httpx</code> não possui suporte nativo a websockets, enquanto <code>aiohttp</code> oferece:</p>

<pre><code class="language-python">import aiohttp

import asyncio

async def usar_websocket():

async with aiohttp.ClientSession() as sessao:

async with sessao.ws_connect(&quot;wss://echo.websocket.org&quot;) as ws:

Enviar mensagem

await ws.send_str(&quot;Olá, WebSocket!&quot;)

Receber mensagem

msg = await ws.receive()

print(f&quot;Resposta: {msg.data}&quot;)

Este exemplo é ilustrativo; o servidor echo pode não estar disponível

asyncio.run(usar_websocket())</code></pre>

<p>Essa capacidade torna <code>aiohttp</code> indispensável para aplicações que necessitam comunicação bidirecional em tempo real.</p>

<h3>Tratamento de Erros em aiohttp</h3>

<pre><code class="language-python">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(&quot;https://httpbin.org/status/404&quot;) as resposta:

resposta.raise_for_status()

return await resposta.json()

except aiohttp.ClientConnectorError as e:

print(f&quot;Erro de conexão: {e}&quot;)

except aiohttp.ClientPayloadError as e:

print(f&quot;Erro ao ler resposta: {e}&quot;)

except asyncio.TimeoutError:

print(&quot;Requisição expirou&quot;)

asyncio.run(requisicao_segura_aiohttp())</code></pre>

<h2>httpx vs aiohttp: Qual Escolher?</h2>

<p>Ambas as bibliotecas são maduras e confiáveis. A escolha depende do contexto do seu projeto. Use <code>httpx</code> se você precisa apenas de requisições HTTP simples e limpas — sua API é mais intuitiva e documentation melhor. Use <code>aiohttp</code> se você já está familiarizado com ela, precisa de websockets nativos, ou está em um projeto legado que já a utiliza.</p>

<h3>Comparação Prática</h3>

<pre><code class="language-python"># httpx: mais simples, mais moderno

import httpx

import asyncio

async def httpx_exemplo():

async with httpx.AsyncClient() as client:

r = await client.get(&quot;https://api.example.com/data&quot;)

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(&quot;https://api.example.com/data&quot;) as r:

return await r.json()

Ambas fazem a mesma coisa, mas com sintaxes diferentes</code></pre>

<p>Para novos projetos recomendo <code>httpx</code>: melhor documentação, API mais consistente com <code>requests</code> que você provavelmente conhece, e suporte mais ativo. Para websockets, não há escolha — <code>aiohttp</code> é necessário.</p>

<h2>Patterns Avançados</h2>

<h3>Pool de Conexões e Reutilização</h3>

<p>Em aplicações de longa duração, reutilize o <code>AsyncClient</code> ou <code>ClientSession</code>:</p>

<pre><code class="language-python">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(&quot;https://api.example.com/dados&quot;)

print(resposta.json())

finally:

await api.disconnect()

asyncio.run(main())</code></pre>

<p>Isso evita criar e destruir clients repetidamente, reduzindo overhead.</p>

<h3>Semaphore: Limitando Concorrência</h3>

<p>Quando você tem muitas requisições, limitar concorrência evita sobrecarregar o servidor ou sua máquina:</p>

<pre><code class="language-python">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&quot;https://api.example.com/item/{i}&quot; for i in range(100)]

asyncio.run(requisicoes_com_limite(urls))</code></pre>

<h3>Retry com Backoff Exponencial</h3>

<p>Requisições falham. Implementar retry inteligente é essencial:</p>

<pre><code class="language-python">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&quot;Tentativa {tentativa + 1} falhou. Aguardando {espera}s...&quot;)

await asyncio.sleep(espera)

Uso

try:

resposta = asyncio.run(requisicao_com_retry(&quot;https://api.example.com/unstable&quot;))

except Exception as e:

print(f&quot;Falhou após retries: {e}&quot;)</code></pre>

<h2>Conclusão</h2>

<p>Programação assíncrona com <code>httpx</code> e <code>aiohttp</code> transforma a eficiência de aplicações que lidam com requisições HTTP. O ponto fundamental é compreender que <code>async/await</code> permite que múltiplas operações de I/O rodem concorrentemente sem threads, usando um único event loop para orquestração. <code>httpx</code> é a escolha moderna e recomendada para a maioria dos casos por sua API limpa e intuitiva; <code>aiohttp</code> 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.</p>

<h2>Referências</h2>

<ul>

<li><a href="https://www.python-httpx.org/" target="_blank" rel="noopener noreferrer">httpx - Official Documentation</a></li>

<li><a href="https://docs.aiohttp.org/" target="_blank" rel="noopener noreferrer">aiohttp - Official Documentation</a></li>

<li><a href="https://docs.python.org/3/library/asyncio.html" target="_blank" rel="noopener noreferrer">Python asyncio - Official Documentation</a></li>

<li><a href="https://realpython.com/async-io-python/" target="_blank" rel="noopener noreferrer">Real Python - Async IO in Python</a></li>

<li><a href="https://www.datacamp.com/tutorial/making-http-requests-in-python" target="_blank" rel="noopener noreferrer">Using Python Requests and HTTPx for API Interactions</a></li>

</ul>

<p>&lt;!-- FIM --&gt;</p>

Comentários

Mais em Python

Guia Completo de Autenticação em FastAPI: JWT, OAuth2 e Segurança de Endpoints
Guia Completo de Autenticação em FastAPI: JWT, OAuth2 e Segurança de Endpoints

Por que Autenticação é Crítica em APIs Modernas Quando você constrói uma API,...

Dominando Celery em Python: Filas de Tarefas, Workers e Beat Scheduler em Projetos Reais
Dominando Celery em Python: Filas de Tarefas, Workers e Beat Scheduler em Projetos Reais

O que é Celery e por que você precisa disso Celery é uma biblioteca Python as...

Django REST Framework: Serializers, ViewSets e Autenticação: Do Básico ao Avançado
Django REST Framework: Serializers, ViewSets e Autenticação: Do Básico ao Avançado

Serializers: A Base da Transformação de Dados Um serializer no Django REST Fr...