Configurazione del proxy in Python requests e aiohttp: guida completa con esempi di codice

Nello sviluppo di parser, automazione della raccolta dati o test di servizi web, è spesso necessario utilizzare server proxy da Python. Le librerie requests e aiohttp offrono meccanismi flessibili per lavorare con i proxy, ma la loro configurazione presenta importanti sfide. In questa guida esamineremo approcci sincroni e asincroni, mostreremo esempi per proxy HTTP e SOCKS5, discuteremo la rotazione IP e la gestione degli errori.

Configurazione di base del proxy in requests

La libreria requests è lo standard per le richieste HTTP in Python. La configurazione del proxy avviene tramite il parametro proxies, che accetta un dizionario con i protocolli e gli indirizzi dei server proxy.

Esempio semplice con proxy HTTP:

import requests

# Configurazione del proxy
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Esecuzione della richiesta tramite proxy
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Nota: per le richieste HTTPS è necessario specificare anche il protocollo http:// nel valore del proxy (non https://). Questo perché la connessione al server proxy viene stabilita tramite HTTP e poi, tramite il metodo CONNECT, viene creato un tunnel per il traffico HTTPS.

Utilizzo delle variabili d'ambiente:

La libreria requests legge automaticamente i proxy dalle variabili d'ambiente HTTP_PROXY e HTTPS_PROXY:

import os
import requests

# Impostazione tramite variabili d'ambiente
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Il proxy verrà applicato automaticamente
response = requests.get('https://httpbin.org/ip')
print(response.json())

Questo approccio è comodo per la containerizzazione (Docker) o quando i proxy sono configurati a livello di sistema. Tuttavia, per maggiore flessibilità, si raccomanda di passare esplicitamente il parametro proxies.

Autenticazione e SOCKS5 in requests

La maggior parte dei servizi proxy commerciali richiede l'autenticazione con nome utente e password. In requests, questo viene realizzato tramite il formato URL con le credenziali.

Proxy HTTP con autenticazione:

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

Configurazione del proxy SOCKS5:

Per lavorare con SOCKS5 è necessaria una libreria aggiuntiva requests[socks] o PySocks. Installazione:

pip install requests[socks]

Esempio di utilizzo di SOCKS5:

import requests

# SOCKS5 senza autenticazione
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 con autenticazione
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())

I proxy SOCKS5 sono particolarmente utili quando si lavora con proxy residenziali, poiché questo protocollo fornisce un tunneling del traffico più sicuro e supporta UDP (necessario per alcune applicazioni).

Rotazione dei proxy in requests

Durante il parsing di grandi volumi di dati, l'uso di un solo indirizzo IP porta a blocchi. La rotazione dei proxy è il cambio ciclico degli IP per distribuire il carico e aggirare i limiti di richiesta.

Rotazione semplice tramite lista:

import requests
import itertools

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

# Creazione di un iteratore infinito
proxy_pool = itertools.cycle(proxy_list)

# Esecuzione delle richieste con rotazione
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"Richiesta {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Errore con il proxy {proxy}: {e}")

Rotazione con sessioni per mantenere i cookie:

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)

# Utilizzo
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"Richiesta {i+1}: {response.json()['origin']}")

Rotazione casuale per imprevedibilità:

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}

# Ogni richiesta con un proxy casuale
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Richiesta {i+1}: {response.json()['origin']}")

La rotazione casuale è più efficace quando si lavora con siti che monitorano i modelli delle richieste. Il cambio sequenziale degli IP può sembrare sospetto, mentre la scelta casuale imita il comportamento di diversi utenti.

Configurazione del proxy in aiohttp

La libreria aiohttp è progettata per richieste HTTP asincrone ed è critica per parser ad alto carico. La configurazione del proxy è diversa da requests: si utilizza il parametro proxy (singolare).

Esempio di base 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)

# Esecuzione
asyncio.run(fetch_with_proxy())

Proxy con autenticazione:

In aiohttp, l'autenticazione viene passata tramite l'oggetto aiohttp.BasicAuth o direttamente nell'URL:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Opzione 1: Credenziali nell'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())

# Opzione 2: Tramite BasicAuth (per alcuni proxy)
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 in aiohttp:

Per SOCKS5 è necessaria la libreria 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())

Quando si lavora con proxy mobili per il parsing di social network o marketplace, è consigliabile utilizzare aiohttp: l'asincronicità consente di gestire centinaia di richieste in parallelo senza bloccare il flusso di esecuzione.

Rotazione asincrona e pool di proxy

Per parser ad alto carico, è fondamentale una rotazione efficace dei proxy con gestione dei guasti e sostituzione automatica degli IP non funzionanti. Esaminiamo schemi avanzati per aiohttp.

Classe per gestire il pool di proxy:

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]:
        """Ottenere il prossimo proxy funzionante"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Tutti i proxy non disponibili
    
    def mark_failed(self, proxy: str):
        """Contrassegnare il proxy come non funzionante"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} contrassegnato come non disponibile")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Eseguire la richiesta con cambio automatico del proxy in caso di errore"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Tutti i proxy non disponibili")
            
            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"Errore con il proxy {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Impossibile eseguire la richiesta dopo {max_retries} tentativi")

# Utilizzo
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:
        # Esecuzione di 10 richieste con rotazione automatica
        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"Richiesta {i+1} terminata con errore: {result}")
            else:
                print(f"Richiesta {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Elaborazione parallela con limitazione della concorrenza:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Limitazione delle richieste simultanee
        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 richieste
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Limitazione: non più di 10 richieste simultanee
    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)
        
        # Analisi dei risultati
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Richieste riuscite: {len(successful)}")
        print(f"Richieste non riuscite: {len(failed)}")

asyncio.run(main())

L'uso di asyncio.Semaphore è critico quando si lavora con i proxy: un numero eccessivo di connessioni simultanee tramite un solo IP può causare blocchi da parte del sito di destinazione o del fornitore di proxy.

Gestione degli errori e dei timeout

Lavorare con i proxy comporta un numero elevato di errori: timeout, interruzioni della connessione, fallimenti dei server proxy. Una corretta gestione degli errori è la chiave per la stabilità del parser.

Errori tipici durante l'uso dei proxy:

Gestione degli errori in 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 di connessione, timeout di lettura)
                allow_redirects=True,
                verify=True  # Verifica del certificato SSL
            )
            response.raise_for_status()  # Solleverà un'eccezione per 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Tentativo {attempt + 1}: Proxy non disponibile - {e}")
        except ConnectTimeout as e:
            print(f"Tentativo {attempt + 1}: Timeout di connessione - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Tentativo {attempt + 1}: Errore HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Errore di autenticazione del proxy!")
                break  # Non ripetere in caso di errore di autenticazione
        except RequestException as e:
            print(f"Tentativo {attempt + 1}: Errore generale - {e}")
        
        if attempt < max_retries - 1:
            print(f"Riprova tra 2 secondi...")
            import time
            time.sleep(2)
    
    raise Exception(f"Impossibile eseguire la richiesta dopo {max_retries} tentativi")

# Utilizzo
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"Errore critico: {e}")

Gestione degli errori in 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"Tentativo {attempt + 1}: Errore di connessione al proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Tentativo {attempt + 1}: Timeout")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Tentativo {attempt + 1}: Errore HTTP del proxy - {e}")
            if e.status == 407:
                print("Errore di autenticazione del proxy!")
                break
        except ClientError as e:
            print(f"Tentativo {attempt + 1}: Errore generale del client - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Impossibile eseguire la richiesta dopo {max_retries} tentativi")

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"Errore critico: {e}")

asyncio.run(main())

Impostazione dei timeout:

Una corretta impostazione dei timeout è fondamentale per la stabilità. Valori raccomandati:

• Timeout di connessione: 5-10 secondi (tempo per stabilire la connessione con il proxy)
• Timeout di lettura: 10-30 secondi (tempo per ricevere una risposta dal sito di destinazione)
• Timeout totale: 30-60 secondi (tempo totale della richiesta)

Per proxy residenziali lenti, è consigliabile aumentare i timeout a 20-30 secondi per connessione, poiché il routing tramite fornitori reali può richiedere più tempo.

Migliori pratiche e ottimizzazione

Un lavoro efficace con i proxy richiede il rispetto di un insieme di regole per minimizzare i blocchi e massimizzare le prestazioni.

1. Utilizzo di Session per riutilizzare le connessioni:

# requests: Session riutilizza le connessioni TCP
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Più veloce di requests.get()

# aiohttp: Session è obbligatoria per l'asincronicità
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Impostazione di User-Agent e intestazioni realistiche:

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. Limitazione del rate limit (richieste al secondo):

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

# Utilizzo: non più di 2 richieste al secondo
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

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

4. Logging e monitoraggio dei proxy:

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
        
        # Logging ogni 10 richieste
        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} richieste, successo {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# Nel codice della richiesta
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"Errore con il proxy {proxy}: {e}")

5. Cache DNS per velocizzare:

# aiohttp con caching 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:
    # Le richieste utilizzeranno la cache DNS per 5 minuti
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Gestione di captcha e blocchi:

Confronto tra requests e aiohttp per i proxy

La scelta tra requests e aiohttp dipende dal compito e dal volume di dati. Esaminiamo le differenze chiave.

Quando utilizzare requests:

• Script semplici per compiti una tantum
• Prototipazione e testing
• Basso volume di richieste (fino a 100 al minuto)
• Quando la semplicità del codice e la leggibilità sono importanti
• Integrazione con librerie sincrone

Quando utilizzare aiohttp:

• Parsing di grandi volumi di dati (migliaia di pagine)
• Monitoraggio di molteplici fonti in tempo reale
• Servizi API con alto carico
• Quando la velocità di elaborazione è critica
• Lavorare con WebSocket tramite proxy

Confronto pratico delle prestazioni:

# Test: 100 richieste tramite proxy

# requests (sincrono) - ~50 secondi
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} secondi")

# aiohttp (asincrono) - ~5 secondi
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} secondi")

Quando si utilizzano proxy di data center per il parsing ad alta velocità, aiohttp mostra un vantaggio di 10-20 volte rispetto a requests grazie all'elaborazione parallela delle richieste.

Conclusione

La configurazione dei proxy in Python tramite le librerie requests e aiohttp è un'abilità fondamentale per lo sviluppo di parser, automazione della raccolta dati e aggiramento delle restrizioni geografiche. La libreria requests è adatta per script semplici e prototipazione grazie a un'API sincrona chiara, mentre aiohttp offre alte prestazioni nella gestione di migliaia di richieste tramite un'architettura asincrona.

Punti chiave per un lavoro efficace con i proxy in Python: corretta gestione degli errori e dei timeout, implementazione della rotazione degli indirizzi IP per distribuire il carico, utilizzo di Session per riutilizzare le connessioni, impostazione di intestazioni e User-Agent realistici, monitoraggio delle prestazioni dei server proxy. Per i proxy SOCKS5 sono necessarie librerie aggiuntive: requests[socks] o aiohttp-socks.

Quando si sceglie il tipo di proxy per il parsing, considerare la specificità del compito: per parser ad alto carico con migliaia di richieste, utilizzare proxy di data center veloci; per aggirare sistemi anti-bot rigidi e lavorare con social network, si raccomandano proxy residenziali con IP reali di utenti domestici; per compiti che richiedono la massima anonimato e simulazione del traffico mobile, i proxy mobili con IP di operatori di telefonia mobile sono ottimali.

Se si prevede di sviluppare parser ad alte prestazioni o automatizzare la raccolta di dati da molteplici fonti, si consiglia di provare proxy residenziali: offrono un alto livello di anonimato, minimo rischio di blocchi e funzionamento stabile con la maggior parte dei servizi web protetti. Per compiti tecnici con alta velocità di elaborazione, sono adatti anche proxy di data center con bassa latenza e alta capacità di banda.