Configuração de proxy em Python requests e aiohttp: guia completo com exemplos de código

Ao desenvolver parsers, automatizar a coleta de dados ou testar serviços web em Python, muitas vezes é necessário usar servidores proxy. As bibliotecas requests e aiohttp fornecem mecanismos flexíveis para trabalhar com proxies, mas sua configuração tem nuances importantes. Neste guia, vamos explorar abordagens síncronas e assíncronas, mostrar exemplos para proxies HTTP e SOCKS5, discutir rotação de IP e tratamento de erros.

Configuração básica de proxy em requests

A biblioteca requests é o padrão para requisições HTTP em Python. A configuração de proxy é feita através do parâmetro proxies, que aceita um dicionário com protocolos e endereços de servidores proxy.

Exemplo simples com proxy HTTP:

import requests

# Configuração do proxy
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Execução da requisição através do proxy
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Observe: para requisições HTTPS, o protocolo http:// também deve ser especificado no valor do proxy (não https://). Isso se deve ao fato de que a conexão com o servidor proxy é estabelecida via HTTP, e então, através do método CONNECT, um túnel é criado para o tráfego HTTPS.

Uso de variáveis de ambiente:

A biblioteca requests lê automaticamente os proxies das variáveis de ambiente HTTP_PROXY e HTTPS_PROXY:

import os
import requests

# Configuração através de variáveis de ambiente
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Proxy será aplicado automaticamente
response = requests.get('https://httpbin.org/ip')
print(response.json())

Essa abordagem é conveniente para containerização (Docker) ou quando os proxies são configurados a nível de sistema. No entanto, para flexibilidade, é recomendada a passagem explícita do parâmetro proxies.

Autenticação e SOCKS5 em requests

A maioria dos serviços de proxy comerciais requer autenticação por login e senha. Em requests, isso é implementado através do formato de URL com as credenciais.

Proxy HTTP com autenticação:

import requests

# Formato: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

Configuração de proxy SOCKS5:

Para trabalhar com SOCKS5, é necessária uma biblioteca adicional requests[socks] ou PySocks. Instalação:

pip install requests[socks]

Exemplo de uso de SOCKS5:

import requests

# SOCKS5 sem autenticação
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 com autenticação
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

Proxies SOCKS5 são especialmente úteis ao trabalhar com proxies residenciais, pois esse protocolo oferece um tunelamento de tráfego mais confiável e suporta UDP (necessário para alguns aplicativos).

Rotação de proxies em requests

Ao fazer parsing de grandes volumes de dados, o uso de um único endereço IP leva a bloqueios. A rotação de proxies é a troca cíclica de IPs para distribuir a carga e contornar limites de taxa.

Rotação simples através de uma lista:

import requests
import itertools

# Lista de servidores proxy
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# Criação de um iterador infinito
proxy_pool = itertools.cycle(proxy_list)

# Execução de requisições com rotação
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"Requisição {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Erro com o proxy {proxy}: {e}")

Rotação com sessões para manter cookies:

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# Uso
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"Requisição {i+1}: {response.json()['origin']}")

Rotação aleatória para imprevisibilidade:

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# Cada requisição com proxy aleatório
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Requisição {i+1}: {response.json()['origin']}")

A rotação aleatória é mais eficaz ao trabalhar com sites que monitoram padrões de requisições. A troca sequencial de IP pode parecer suspeita, enquanto a escolha aleatória imita o comportamento de diferentes usuários.

Configuração de proxy em aiohttp

A biblioteca aiohttp é projetada para requisições HTTP assíncronas e é crítica para parsers de alta carga. A configuração de proxy difere de requests — utiliza o parâmetro proxy (no singular).

Exemplo básico com proxy HTTP:

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = 'http://123.45.67.89:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            data = await response.json()
            print(data)

# Execução
asyncio.run(fetch_with_proxy())

Proxy com autenticação:

Em aiohttp, a autenticação é passada através do objeto aiohttp.BasicAuth ou diretamente na URL:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Opção 1: Credenciais na URL
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# Opção 2: Através do BasicAuth (para alguns proxies)
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

SOCKS5 em aiohttp:

Para SOCKS5, é necessária a biblioteca aiohttp-socks:

pip install aiohttp-socks

import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

Ao trabalhar com proxies móveis para parsing de redes sociais ou marketplaces, é recomendável usar aiohttp — a assíncrona permite processar centenas de requisições em paralelo sem bloquear o fluxo de execução.

Rotação assíncrona e pool de proxies

Para parsers de alta carga, é crítica uma rotação eficaz de proxies com tratamento de falhas e substituição automática de IPs não funcionais. Vamos explorar padrões avançados para aiohttp.

Classe para gerenciar o pool de proxies:

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """Obter o próximo proxy funcional"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Todos os proxies estão indisponíveis
    
    def mark_failed(self, proxy: str):
        """Marcar proxy como não funcional"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} marcado como indisponível")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Executar requisição com troca automática de proxy em caso de erro"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Todos os proxies estão indisponíveis")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"Erro com o proxy {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Não foi possível executar a requisição após {max_retries} tentativas")

# Uso
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # Execução de 10 requisições com rotação automática
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Requisição {i+1} falhou: {result}")
            else:
                print(f"Requisição {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Processamento paralelo com limitação de concorrência:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Limitação de requisições simultâneas
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50 requisições
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Limitação: não mais que 10 requisições simultâneas
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # Análise dos resultados
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Requisições bem-sucedidas: {len(successful)}")
        print(f"Requisições falhadas: {len(failed)}")

asyncio.run(main())

O uso de asyncio.Semaphore é crítico ao trabalhar com proxies — um número excessivo de conexões simultâneas através de um único IP pode resultar em bloqueios por parte do site alvo ou do provedor de proxy.

Tratamento de erros e timeouts

Trabalhar com proxies está associado a um aumento no número de erros: timeouts, desconexões, falhas de servidores proxy. Um tratamento adequado de erros é a chave para a estabilidade do parser.

Erros típicos ao trabalhar com proxies:

Tratamento de erros em requests:

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (timeout de conexão, timeout de leitura)
                allow_redirects=True,
                verify=True  # Verificação do certificado SSL
            )
            response.raise_for_status()  # Levanta uma exceção em caso de 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Tentativa {attempt + 1}: Proxy indisponível - {e}")
        except ConnectTimeout as e:
            print(f"Tentativa {attempt + 1}: Timeout de conexão - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Tentativa {attempt + 1}: Erro HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Erro de autenticação do proxy!")
                break  # Não repetir em caso de erro de autenticação
        except RequestException as e:
            print(f"Tentativa {attempt + 1}: Erro geral - {e}")
        
        if attempt < max_retries - 1:
            print(f"Repetindo em 2 segundos...")
            import time
            time.sleep(2)
    
    raise Exception(f"Não foi possível executar a requisição após {max_retries} tentativas")

# Uso
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"Erro crítico: {e}")

Tratamento de erros em aiohttp:

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"Tentativa {attempt + 1}: Erro de conexão com o proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Tentativa {attempt + 1}: Timeout")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Tentativa {attempt + 1}: Erro HTTP do proxy - {e}")
            if e.status == 407:
                print("Erro de autenticação do proxy!")
                break
        except ClientError as e:
            print(f"Tentativa {attempt + 1}: Erro geral do cliente - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Não foi possível executar a requisição após {max_retries} tentativas")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"Erro crítico: {e}")

asyncio.run(main())

Configuração de timeouts:

A configuração adequada de timeouts é crítica para a estabilidade. Valores recomendados:

• Timeout de conexão: 5-10 segundos (tempo para estabelecer conexão com o proxy)
• Timeout de leitura: 10-30 segundos (tempo para receber resposta do site alvo)
• Timeout total: 30-60 segundos (tempo total da requisição)

Para proxies residenciais lentos, recomenda-se aumentar os timeouts para 20-30 segundos por conexão, pois a roteação através de provedores reais pode levar mais tempo.

Melhores práticas e otimização

Trabalhar de forma eficaz com proxies requer seguir um conjunto de regras para minimizar bloqueios e maximizar desempenho.

1. Usar Session para reutilizar conexões:

# requests: Session reutiliza conexões TCP
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Mais rápido que requests.get()

# aiohttp: Session é obrigatória para assíncrono
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Configurar User-Agent e cabeçalhos realistas:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. Limitar a taxa de requisições (requisições por segundo):

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# Uso: não mais que 2 requisições por segundo
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. Registro e monitoramento de proxies:

import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # Registro a cada 10 requisições
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"Proxy {proxy}: {total} requisições, sucesso {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# No código de requisição
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"Erro com o proxy {proxy}: {e}")

5. Cache de DNS para aceleração:

# aiohttp com cache de DNS
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # As requisições usarão o cache de DNS por 5 minutos
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Tratamento de CAPTCHA e bloqueios:

Comparação entre requests e aiohttp para proxies

A escolha entre requests e aiohttp depende da tarefa e do volume de dados. Vamos considerar as principais diferenças.

Quando usar requests:

• Scripts simples para tarefas únicas
• Prototipagem e testes
• Baixo volume de requisições (até 100 por minuto)
• Quando a simplicidade do código e legibilidade são importantes
• Integração com bibliotecas síncronas

Quando usar aiohttp:

• Parsing de grandes volumes de dados (milhares de páginas)
• Monitoramento de múltiplas fontes em tempo real
• Serviços de API com alta carga
• Quando a velocidade de processamento é crítica
• Trabalho com WebSocket através de proxies

Comparação prática de desempenho:

# Teste: 100 requisições através de proxy

# requests (síncrono) - ~50 segundos
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} segundos")

# aiohttp (assíncrono) - ~5 segundos
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} segundos")

Ao usar proxies de datacenter para parsing em alta velocidade, aiohttp mostra uma vantagem de 10-20 vezes em comparação com requests devido ao processamento paralelo das requisições.

Conclusão

A configuração de proxies em Python através das bibliotecas requests e aiohttp é uma habilidade fundamental para o desenvolvimento de parsers, automação de coleta de dados e contorno de restrições geográficas. A biblioteca requests é adequada para scripts simples e prototipagem devido à sua API síncrona clara, enquanto aiohttp proporciona alto desempenho ao processar milhares de requisições através de uma arquitetura assíncrona.

Pontos-chave para um trabalho eficaz com proxies em Python: tratamento adequado de erros e timeouts, implementação de rotação de endereços IP para distribuição de carga, uso de Session para reutilização de conexões, configuração de cabeçalhos e User-Agent realistas, monitoramento do desempenho dos servidores proxy. Para proxies SOCKS5, são necessárias bibliotecas adicionais — requests[socks] ou aiohttp-socks.

Ao escolher o tipo de proxy para parsing, considere a especificidade da tarefa: para parsers de alta carga com milhares de requisições, proxies de datacenter rápidos são adequados; para contornar sistemas anti-bot rigorosos e trabalhar com redes sociais, recomenda-se proxies residenciais com IPs reais de usuários domésticos; e para tarefas que exigem máxima anonimidade e simulação de tráfego móvel, proxies móveis com IPs de operadoras de telefonia celular são ideais.

Se você planeja desenvolver parsers de alto desempenho ou automatizar a coleta de dados de várias fontes, recomendamos experimentar proxies residenciais — eles oferecem alto nível de anonimato, risco mínimo de bloqueios e operação estável com a maioria dos serviços web protegidos. Para tarefas técnicas com alta velocidade de processamento, também são adequados proxies de datacenter com baixa latência e alta largura de banda.