Usando proxies em Python com aiohttp

Usando proxies em Python com aiohttp

aiohttp e proxies

aiohttp é a biblioteca HTTP assíncrona mais popular do Python. Ela permite enviar milhares de requisições concorrentes, o que a torna ideal para parsing e automação. O suporte a proxies no aiohttp é um recurso essencial em tarefas de larga escala.

Uso básico de proxy

Proxies HTTP

A forma mais simples de usar um proxy HTTP no aiohttp é com o parâmetro proxy no método de requisição:

import aiohttp
import asyncio

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

asyncio.run(fetch_with_proxy())

Proxies com autenticação

Para proxies com usuário e senha, use BasicAuth:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    proxy = "http://proxy_ip:port"
    proxy_auth = aiohttp.BasicAuth("username", "password")

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy=proxy,
            proxy_auth=proxy_auth
        ) as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_auth_proxy())

Um formato alternativo é colocar as credenciais na URL:

proxy = "http://username:password@proxy_ip:port"

Proxies SOCKS5

Instalando o aiohttp-socks

O aiohttp não tem suporte nativo a SOCKS. Instale o aiohttp-socks:

pip install aiohttp-socks

Uso

import aiohttp
from aiohttp_socks import ProxyConnector

async def fetch_with_socks():
    connector = ProxyConnector.from_url("socks5://user:pass@proxy_ip:port")

    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_socks())

Protocolos suportados

O ProxyConnector suporta:
- socks5://
- socks4://
- http://
- https://

Rotação de proxies

Rotação simples

import aiohttp
import asyncio
import random

PROXIES = [
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
    "http://user:pass@proxy3:port",
]

async def fetch_with_rotation(url):
    proxy = random.choice(PROXIES)
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy=proxy) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    tasks = [fetch_with_rotation(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"OK: {len(r)} bytes")

asyncio.run(main())

Rotação com exclusão de proxies com falha

import aiohttp
import asyncio
from collections import deque

class ProxyRotator:
    def __init__(self, proxies):
        self.proxies = deque(proxies)
        self.failed = set()

    def get_proxy(self):
        for _ in range(len(self.proxies)):
            proxy = self.proxies[0]
            self.proxies.rotate(-1)
            if proxy not in self.failed:
                return proxy
        raise Exception("All proxies failed")

    def mark_failed(self, proxy):
        self.failed.add(proxy)

    def mark_success(self, proxy):
        self.failed.discard(proxy)

async def fetch(session, url, rotator, retries=3):
    for attempt in range(retries):
        proxy = rotator.get_proxy()
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    rotator.mark_success(proxy)
                    return await resp.text()
                elif resp.status == 403:
                    rotator.mark_failed(proxy)
        except Exception:
            rotator.mark_failed(proxy)
    return None

Requisições concorrentes com limites

Semáforo para controle de concorrência

import aiohttp
import asyncio

async def fetch(session, url, proxy, semaphore):
    async with semaphore:
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=15)) as resp:
                return await resp.text()
        except Exception as e:
            return None

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(100)]
    proxy = "http://user:pass@proxy:port"
    semaphore = asyncio.Semaphore(10)  # máx. 10 simultâneas

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, proxy, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)
        success = sum(1 for r in results if r)
        print(f"Success: {success}/{len(urls)}")

asyncio.run(main())

Configuração da sessão

Timeout

timeout = aiohttp.ClientTimeout(
    total=30,        # timeout total
    connect=10,      # timeout de conexão
    sock_read=10     # timeout de leitura do socket
)
session = aiohttp.ClientSession(timeout=timeout)

Headers

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "en-US,en;q=0.9",
}
session = aiohttp.ClientSession(headers=headers)

SSL

# Desativando a verificação SSL (para testes)
import ssl
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl_context))

Tratamento de erros

import aiohttp

async def safe_fetch(session, url, proxy):
    try:
        async with session.get(url, proxy=proxy) as resp:
            if resp.status == 200:
                return await resp.text()
            elif resp.status == 407:
                print("Proxy authentication required")
            elif resp.status == 403:
                print("Forbidden - proxy might be banned")
            elif resp.status == 429:
                print("Rate limited - slow down")
            return None
    except aiohttp.ClientProxyConnectionError:
        print("Cannot connect to proxy")
    except aiohttp.ClientConnectorError:
        print("Connection error")
    except asyncio.TimeoutError:
        print("Request timed out")
    except Exception as e:
        print(f"Unexpected error: {e}")
    return None

aiohttp vs requests

Conclusão

aiohttp é a melhor escolha para uso assíncrono de proxies em Python. O suporte a HTTP e SOCKS5 (via aiohttp-socks), a rotação de proxies, o controle de concorrência e o tratamento de erros fazem dela uma ferramenta poderosa para parsing e automação em larga escala.