Configuración de proxies en Python requests y aiohttp: guía completa con ejemplos de código

Al desarrollar parsers, automatizando la recolección de datos o probando servicios web desde Python, a menudo es necesario utilizar servidores proxy. Las bibliotecas requests y aiohttp proporcionan mecanismos flexibles para trabajar con proxies, pero su configuración tiene matices importantes. En esta guía, analizaremos enfoques sincrónicos y asincrónicos, mostraremos ejemplos para proxies HTTP y SOCKS5, revisaremos la rotación de IP y el manejo de errores.

Configuración básica de proxy en requests

La biblioteca requests es el estándar para solicitudes HTTP en Python. La configuración de proxy se realiza a través del parámetro proxies, que acepta un diccionario con protocolos y direcciones de servidores proxy.

Ejemplo más simple con proxy HTTP:

import requests

# Configuración del proxy
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Realizando una solicitud a través del proxy
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Nota: Para solicitudes HTTPS, también se indica el protocolo http:// en el valor del proxy (no https://). Esto se debe a que la conexión con el servidor proxy se establece a través de HTTP, y luego se crea un túnel para el tráfico HTTPS mediante el método CONNECT.

Uso de variables de entorno:

La biblioteca requests lee automáticamente los proxies de las variables de entorno HTTP_PROXY y HTTPS_PROXY:

import os
import requests

# Configuración a través de variables de entorno
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# El proxy se aplicará automáticamente
response = requests.get('https://httpbin.org/ip')
print(response.json())

Este enfoque es conveniente para la contenedorización (Docker) o cuando los proxies se configuran a nivel del sistema. Sin embargo, para mayor flexibilidad, se recomienda la transmisión explícita del parámetro proxies.

Autenticación y SOCKS5 en requests

La mayoría de los servicios de proxy comerciales requieren autenticación mediante nombre de usuario y contraseña. En requests, esto se implementa a través del formato de URL con las credenciales.

Proxy HTTP con autenticación:

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())

Configuración de proxy SOCKS5:

Para trabajar con SOCKS5 se requiere una biblioteca adicional requests[socks] o PySocks. Instalación:

pip install requests[socks]

Ejemplo de uso de SOCKS5:

import requests

# SOCKS5 sin autenticación
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 con autenticación
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())

Los proxies SOCKS5 son especialmente útiles al trabajar con proxies residenciales, ya que este protocolo proporciona un túnel de tráfico más confiable y admite UDP (necesario para algunas aplicaciones).

Rotación de proxies en requests

Al analizar grandes volúmenes de datos, el uso de una sola dirección IP puede llevar a bloqueos. La rotación de proxies es un cambio cíclico de IP para distribuir la carga y eludir los límites de tasa.

Rotación simple a través de una 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',
]

# Creación de un iterador infinito
proxy_pool = itertools.cycle(proxy_list)

# Realizando solicitudes con rotación
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"Solicitud {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Error con el proxy {proxy}: {e}")

Rotación con sesiones para mantener 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"Solicitud {i+1}: {response.json()['origin']}")

Rotación aleatoria para imprevisibilidad:

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 solicitud con un proxy aleatorio
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Solicitud {i+1}: {response.json()['origin']}")

La rotación aleatoria es más efectiva al trabajar con sitios que rastrean patrones de solicitudes. El cambio secuencial de IP puede parecer sospechoso, mientras que la selección aleatoria imita el comportamiento de diferentes usuarios.

Configuración de proxy en aiohttp

La biblioteca aiohttp está diseñada para solicitudes HTTP asíncronas y es crítica para parsers de alta carga. La configuración de proxy es diferente de requests: se utiliza el parámetro proxy (en singular).

Ejemplo básico con 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)

# Ejecución
asyncio.run(fetch_with_proxy())

Proxy con autenticación:

En aiohttp, la autenticación se pasa a través del objeto aiohttp.BasicAuth o directamente en la URL:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Opción 1: Credenciales en la 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())

# Opción 2: A través de BasicAuth (para algunos 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 en aiohttp:

Para SOCKS5 se requiere la 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())

Al trabajar con proxies móviles para analizar redes sociales o marketplaces, se recomienda usar aiohttp: la asincronía permite manejar cientos de solicitudes en paralelo sin bloquear el flujo de ejecución.

Rotación asíncrona y pool de proxies

Para parsers de alta carga, es crítica una rotación efectiva de proxies con manejo de fallos y reemplazo automático de IPs no funcionales. Revisemos patrones avanzados para aiohttp.

Clase para manejar un 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]:
        """Obtener el siguiente 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 los proxies están inactivos
    
    def mark_failed(self, proxy: str):
        """Marcar el proxy como no funcional"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} marcado como inactivo")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Realizar una solicitud con cambio automático de proxy en caso de error"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Todos los proxies están inactivos")
            
            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"Error con el proxy {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"No se pudo realizar la solicitud después de {max_retries} intentos")

# 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:
        # Realizando 10 solicitudes con rotación 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"Solicitud {i+1} finalizó con error: {result}")
            else:
                print(f"Solicitud {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Procesamiento paralelo con limitación de concurrencia:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Limitación de solicitudes 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 solicitudes
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Limitación: no más de 10 solicitudes 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álisis de resultados
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Solicitudes exitosas: {len(successful)}")
        print(f"Solicitudes fallidas: {len(failed)}")

asyncio.run(main())

El uso de asyncio.Semaphore es crítico al trabajar con proxies: un número excesivo de conexiones simultáneas a través de una sola IP puede causar bloqueos por parte del sitio objetivo o del proveedor de proxies.

Manejo de errores y timeouts

Trabajar con proxies conlleva un aumento en la cantidad de errores: timeouts, desconexiones, fallos de servidores proxy. Un manejo adecuado de errores es clave para la estabilidad del parser.

Errores típicos al trabajar con proxies:

Manejo de errores en 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 conexión, timeout de lectura)
                allow_redirects=True,
                verify=True  # Verificación del certificado SSL
            )
            response.raise_for_status()  # Lanzará una excepción en caso de 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Intento {attempt + 1}: Proxy no disponible - {e}")
        except ConnectTimeout as e:
            print(f"Intento {attempt + 1}: Timeout de conexión - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Intento {attempt + 1}: Error HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("¡Error de autenticación del proxy!")
                break  # No repetir en caso de error de autorización
        except RequestException as e:
            print(f"Intento {attempt + 1}: Error general - {e}")
        
        if attempt < max_retries - 1:
            print(f"Reintentando en 2 segundos...")
            import time
            time.sleep(2)
    
    raise Exception(f"No se pudo realizar la solicitud después de {max_retries} intentos")

# 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"Error crítico: {e}")

Manejo de errores en 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"Intento {attempt + 1}: Error de conexión al proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Intento {attempt + 1}: Timeout")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Intento {attempt + 1}: Error HTTP del proxy - {e}")
            if e.status == 407:
                print("¡Error de autenticación del proxy!")
                break
        except ClientError as e:
            print(f"Intento {attempt + 1}: Error general del cliente - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"No se pudo realizar la solicitud después de {max_retries} intentos")

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"Error crítico: {e}")

asyncio.run(main())

Configuración de timeouts:

La configuración adecuada de timeouts es crítica para la estabilidad. Valores recomendados:

• Timeout de conexión: 5-10 segundos (tiempo para establecer conexión con el proxy)
• Timeout de lectura: 10-30 segundos (tiempo para recibir respuesta del sitio objetivo)
• Timeout total: 30-60 segundos (tiempo total de la solicitud)

Para proxies residenciales lentos, se recomienda aumentar los timeouts a 20-30 segundos por conexión, ya que la ruta a través de proveedores reales puede tardar más tiempo.

Mejores prácticas y optimización

Un trabajo efectivo con proxies requiere seguir un conjunto de reglas para minimizar bloqueos y maximizar el rendimiento.

1. Uso de Session para reutilizar conexiones:

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

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

# aiohttp: Session es obligatoria para la asincronía
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Establecer User-Agent y encabezados 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 el rate limit (solicitudes 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: no más de 2 solicitudes por segundo
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

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

4. Registro y monitoreo 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 cada 10 solicitudes
        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} solicitudes, éxito {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# En el código de solicitud
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"Error con el proxy {proxy}: {e}")

5. Caché DNS para acelerar:

# aiohttp con caché 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:
    # Las solicitudes usarán la caché DNS durante 5 minutos
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Manejo de CAPTCHA y bloqueos:

Comparación de requests y aiohttp para proxies

La elección entre requests y aiohttp depende de la tarea y el volumen de datos. Revisemos las diferencias clave.

Cuándo usar requests:

• Scripts simples para tareas puntuales
• Prototipado y pruebas
• Pequeño volumen de solicitudes (hasta 100 por minuto)
• Cuando la simplicidad del código y la legibilidad son importantes
• Integración con bibliotecas sincrónicas

Cuándo usar aiohttp:

• Análisis de grandes volúmenes de datos (miles de páginas)
• Monitoreo de múltiples fuentes en tiempo real
• Servicios API con alta carga
• Cuando la velocidad de procesamiento es crítica
• Trabajo con WebSocket a través de proxies

Comparación práctica del rendimiento:

# Prueba: 100 solicitudes a través de un proxy

# requests (sincrónico) - ~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 (así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")

Al usar proxies de centros de datos para análisis a alta velocidad, aiohttp muestra una ventaja de 10-20 veces en comparación con requests gracias al procesamiento paralelo de solicitudes.

Conclusión

La configuración de proxies en Python a través de las bibliotecas requests y aiohttp es una habilidad fundamental para el desarrollo de parsers, la automatización de la recolección de datos y la elusión de restricciones geográficas. La biblioteca requests es adecuada para scripts simples y prototipado gracias a su API sincrónica clara, mientras que aiohttp proporciona un alto rendimiento al procesar miles de solicitudes a través de una arquitectura asíncrona.

Puntos clave para un trabajo efectivo con proxies en Python: manejo adecuado de errores y timeouts, implementación de rotación de direcciones IP para distribuir la carga, uso de Session para reutilizar conexiones, configuración de encabezados y User-Agent realistas, monitoreo del rendimiento de los servidores proxy. Para proxies SOCKS5 se requieren bibliotecas adicionales: requests[socks] o aiohttp-socks.

Al elegir el tipo de proxy para análisis, considere la especificidad de la tarea: para parsers de alta carga con miles de solicitudes, son adecuados los proxies de centros de datos rápidos; para eludir sistemas anti-bot estrictos y trabajar con redes sociales, se recomiendan proxies residenciales con IPs reales de usuarios domésticos; y para tareas que requieren la máxima anonimidad y simulación de tráfico móvil, los proxies móviles con IPs de operadores de telefonía celular son óptimos.

Si planea desarrollar parsers de alto rendimiento o automatizar la recolección de datos de múltiples fuentes, le recomendamos probar proxies residenciales: ofrecen un alto nivel de anonimato, un riesgo mínimo de bloqueos y un funcionamiento estable con la mayoría de los servicios web protegidos. Para tareas técnicas con alta velocidad de procesamiento, también son adecuados proxies de centros de datos con baja latencia y alta capacidad de transmisión.