Raspagem Assíncrona em Python com httpx, aiohttp e Proxies
Assíncrono transforma um scraper lento em rápido — e um rápido em bloqueado, a menos que você configure corretamente proxies, semáforos e timeouts. Aqui está o padrão completo para httpx e aiohttp, com código.
Se seu scraper passa a maior parte do tempo de relógio esperando na rede, assíncrono é o maior ganho de velocidade disponível — e proxies são o que impede que essa velocidade leve você a ser banido. Este guia cobre raspagem assíncrona em Python com httpx e aiohttp mais proxies de ponta a ponta: como cada biblioteca configura um proxy, como limitar a concorrência com um semáforo, como definir timeouts que realmente disparam, e como rodar IPs sem destruir suas sessões. Tudo com código executável e credenciais de espaço reservado que você troca pelas suas.
Por que assíncrono, e onde os proxies se encaixam
requests padrão é bloqueante: cada chamada espera pela resposta antes que a próxima comece. Buscar 500 páginas e você paga 500 idas e voltas consecutivas. Assíncrono as busca simultaneamente em um loop de eventos, então o tempo total colapsa em direção ao pedido único mais lento em vez da soma. O problema: um surto de pedidos simultâneos de um IP é exatamente a assinatura que os sistemas anti-bot observam. A solução não é desacelerar até parar — é espalhar o tráfego por um pool de proxies rotativos e limitar deliberadamente com um semáforo.
Configurar um proxy em httpx (assíncrono)
httpx é o padrão pragmático porque um modelo de cliente faz tanto síncrono quanto assíncrono, e ele fala HTTP/2. Note a API moderna: é um argumento singular proxy= no cliente, não o antigo dicionário proxies= — uma fonte comum de confusão "por que meu proxy é ignorado" após uma atualização. Credenciais vão direto para a URL do proxy.
import asyncio, httpx
PROXY = "http://USER:PASS@gate.quantumproxies.io:8000"
async def fetch(client, url):
r = await client.get(url, timeout=httpx.Timeout(20.0))
return url, r.status_code, r.text
async def main(urls):
async with httpx.AsyncClient(proxy=PROXY, http2=True) as client:
tasks = [fetch(client, u) for u in urls]
return await asyncio.gather(*tasks)
urls = ["https://httpbin.org/ip"] * 5
print(asyncio.run(main(urls)))Um cliente, reutilizado em cada pedido, é o ponto — ele mantém o pool de conexões aquecido para que você pule apertos de mão TLS repetidos através do proxy. Criar um cliente novo por pedido é o erro de desempenho assíncrono mais comum: ele descarta o reuso de conexão e o estado de cookies em cada chamada.
Configurar um proxy em aiohttp (por pedido)
aiohttp é nativo do asyncio e oferece o controle de concorrência mais fino, mas sua convenção de proxy difere: o proxy é passado por pedido em session.get(), não na sessão. Isso é na verdade conveniente para rotação. Duas coisas mordem as pessoas aqui — o User-Agent padrão é literalmente Python/3.x aiohttp/3.x, uma pista óbvia que você deve substituir, e você deve dimensionar explicitamente o pool de conexões.
import aiohttp, asyncio
PROXY = "http://USER:PASS@gate.quantumproxies.io:8000"
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"}
async def fetch(session, url):
timeout = aiohttp.ClientTimeout(total=30, connect=10, sock_read=20)
async with session.get(url, proxy=PROXY, timeout=timeout) as resp:
return url, resp.status, await resp.text()
async def main(urls):
conn = aiohttp.TCPConnector(limit=20, limit_per_host=8)
async with aiohttp.ClientSession(connector=conn, headers=HEADERS) as session:
return await asyncio.gather(*(fetch(session, u) for u in urls))
print(asyncio.run(main(["https://httpbin.org/ip"] * 5)))TCPConnector(limit=20, limit_per_host=8) limita o total de conexões abertas e conexões para qualquer host único — uma primeira linha de cortesia que impede um alvo de absorver todo o seu pool.
Limitar concorrência com um semáforo
Disparar pedidos simultâneos ilimitados é a maneira mais rápida de queimar um pool de proxies e disparar limites de taxa. Um asyncio.Semaphore é o limitador: ele limita quantos pedidos estão em andamento ao mesmo tempo, independentemente de quantas tarefas você enfileira. Defina um limite global, e para trabalhos agressivos um limite por domínio também.
sem = asyncio.Semaphore(15) # never more than 15 requests in flight
async def guarded_fetch(client, url):
async with sem:
return await fetch(client, url)
async def main(urls):
async with httpx.AsyncClient(proxy=PROXY) as client:
return await asyncio.gather(*(guarded_fetch(client, u) for u in urls))O número certo depende da tolerância do alvo e do tamanho do seu pool, não de quão rápido sua máquina pode ir. Comece conservador (10–20), observe sua taxa de bloqueio, e aumente apenas enquanto o sucesso permanecer alto. Em um único IP fixo, mantenha em dígitos únicos.
Timeouts: modele cada fase
Um pedido com proxy falha em mais lugares do que um direto — resolução DNS, conexão ao proxy, configuração de túnel, conexão ao alvo, espera por cabeçalhos e leitura do corpo são todos atrasos distintos. Um único timeout abrangente esconde qual fase travou. ClientTimeout(total=, connect=, sock_read=) do aiohttp e Timeout() do httpx permitem que você os limite separadamente. A única regra sem exceções: nunca emita um pedido sem um timeout, ou uma saída morta travará uma coroutine para sempre e silenciosamente esgotará seu loop de eventos.
Rodar proxies sem quebrar sessões
Existem duas estratégias de rotação e escolher a errada corrompe seus dados. Rotação aleatória por pedido é perfeita para buscas de páginas sem estado. Mas destrói qualquer fluxo que dependa de cookies, login ou localização, porque o segundo pedido cai em um IP diferente do primeiro. A divisão limpa: rode na fronteira da unidade lógica — um IP por segmento de busca ou por conta — e use um gateway rotativo que lhe entregue uma saída nova automaticamente para que seu código nunca gerencie uma lista.
# rotating gateway: one endpoint, new exit IP per request
ROT = "http://USER:PASS@rotating.quantumproxies.io:8000"
# sticky session: same IP for a multi-step flow, tag the session id
STICKY = "http://USER-session-a1b2:PASS@gate.quantumproxies.io:8000"
async def crawl_segment(urls):
async with httpx.AsyncClient(proxy=ROT) as client: # rotates per call
return await asyncio.gather(*(fetch(client, u) for u in urls))Um gateway residencial rotativo é o padrão pragmático para raspagem de alta concorrência: rotação por pedido em mais de 90 milhões de IPs em mais de 200 países, com sessões fixas quando um carrinho ou login precisa da mesma saída por alguns minutos. Se você está pesando residencial contra ISP ou datacenter para o trabalho, nosso guia sobre qual tipo de proxy usar expõe as compensações.
Obtenha um gateway residencial rotativo
Tentativas, recuo e jitter
Saídas mortas e bloqueios transitórios são normais em escala, não excepcionais. Mas tentativas ingênuas pioram as coisas: quando 50 tarefas assíncronas falham ao mesmo tempo e todas tentam novamente imediatamente, você dispara um surto sincronizado que atinge o alvo mais forte que a execução original. Adicione recuo exponencial mais jitter aleatório para que as tentativas se espalhem, limite as tentativas, e rode o IP em caso de falha em vez de reutilizar o queimado.
import random
from httpx import HTTPError
async def robust_fetch(client, url, tries=3):
for attempt in range(tries):
try:
r = await client.get(url, timeout=httpx.Timeout(20.0))
if r.status_code < 400 and looks_real(r.text):
return r
except HTTPError:
pass
# exponential backoff + jitter before the next attempt
await asyncio.sleep((2 ** attempt) + random.uniform(0, 1))
return None
Um 200 não é um sucesso
O bug de raspagem assíncrona mais sutil é tratar HTTP 200 como concluído. Sistemas anti-bot retornam 200 com uma página CAPTCHA, um aviso de acesso negado, um conjunto de resultados vazio ou um desafio JS — então um proxy que marca "sucesso" apenas pelo código de status está silenciosamente alimentando você com páginas bloqueadas. Valide o conteúdo: verifique um elemento conhecido, um comprimento mínimo, ou a ausência de marcadores de desafio antes de confiar em uma resposta. Esse é o portão looks_real() no loop de tentativas acima.
Quando parar de construir a pilha manualmente
O padrão acima — cliente assíncrono, semáforo, timeouts, rotação, validação de conteúdo — lida com a maioria dos alvos de forma limpa. Mas uma vez que um site adiciona Cloudflare, impressão digital TLS ou renderização pesada do lado do cliente, HTTP assíncrono bruto começa a falhar independentemente do seu proxy, porque um aperto de mão TLS em Python não se parece em nada com o do Chrome. Nesse ponto, uma Scraper API que carrega uma impressão digital de navegador real, roda IPs e renderiza JavaScript sob demanda é menos código e uma taxa de sucesso maior do que manter tudo manualmente. Nosso post sobre custo de navegador headless vs HTTP cobre onde essa escalada compensa.
Perguntas frequentes
Como uso um proxy com aiohttp?
Passe a URL do proxy por pedido: session.get(url, proxy="http://user:pass@host:port"). Ao contrário de requests, aiohttp não aceita um dicionário de proxies na sessão. Sempre substitua o User-Agent padrão (Python/3.x aiohttp/3.x é um sinal óbvio de bot) e defina um ClientTimeout para que uma saída morta não trave a coroutine.
httpx ou aiohttp é melhor para raspagem assíncrona?
Opte por httpx como padrão: um cliente funciona síncrono e assíncrono, HTTP/2 está embutido, e o proxy é um único argumento. Escolha aiohttp quando quiser controle máximo de concorrência — limites explícitos de pool de conexões e proxies por pedido se adequam a buscas grandes e rápidas. Ambos são bons; a estratégia de proxy importa mais do que a biblioteca.
Quantos pedidos simultâneos devo executar?
Não tantos quanto sua máquina permite — tantos quanto o alvo e seu pool toleram. Comece com um semáforo de 10–20 em andamento, observe a taxa de bloqueio e erro, e só aumente enquanto o sucesso permanecer alto. Em um único IP fixo, mantenha em dígitos únicos. Rotação de IPs mais ampla permite que você execute maior concorrência total com segurança.
Por que meu scraper assíncrono é bloqueado quando o síncrono não era?
Porque a concorrência concentra o sinal: muitos pedidos simultâneos de um IP é um padrão clássico de bot. Espalhe a carga por um pool rotativo, limite com um semáforo, adicione recuo com jitter em tentativas, e valide o conteúdo da resposta — um 200 ainda pode ser uma página de desafio. Velocidade sem rotação é o que levou você a ser bloqueado.
Esse é o padrão completo assíncrono: escolha um cliente, configure o proxy da maneira certa para ele, limite a concorrência com um semáforo, limite cada fase de timeout, rode na fronteira lógica, e nunca confie em um 200 puro. Acertar a camada de proxy primeiro e a maioria da lista de bloqueios desaparece antes de você atingi-la.