Mettre en place la rotation de proxys en Python : requests, aiohttp, scrapy

Des exemples de code Python fonctionnels pour la rotation de proxys dans les bibliothèques populaires. Logique de retry personnalisée, gestion du pool de proxys, gestion des erreurs, rate limiting.

requests : rotation de base

L'option la plus simple pour des scripts ponctuels. Passez le proxy directement dans l'appel :

import requests
import random

PROXIES = [
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
]

def fetch(url):
    proxy = random.choice(PROXIES)
    response = requests.get(
        url,
        proxies={'http': proxy, 'https': proxy},
        timeout=15
    )
    return response

# Usage
data = fetch('https://api.ipify.org')
print(data.text)  # each request from a new IP

Pour du residential avec une session sticky, passez le token de session dans le nom d'utilisateur :

import secrets
session_id = secrets.token_hex(8)
proxy = f'http://customer-XXX-session-{session_id}:[email protected]:7000'
# now all requests with this session token use the same IP

requests : un pool de proxys avec retry et blacklist

Une classe prête pour la production qui gère les proxys en échec :

import requests
import time
from collections import deque
from threading import Lock

class ProxyPool:
    def __init__(self, proxies, max_failures=3, blacklist_ttl=300):
        self.proxies = deque(proxies)
        self.failures = {}
        self.blacklist = {}
        self.max_failures = max_failures
        self.blacklist_ttl = blacklist_ttl
        self.lock = Lock()

    def get(self):
        with self.lock:
            now = time.time()
            # clear the blacklist
            self.blacklist = {p: t for p, t in self.blacklist.items()
                              if now - t < self.blacklist_ttl}
            # find a free one
            for _ in range(len(self.proxies)):
                proxy = self.proxies[0]
                self.proxies.rotate(-1)
                if proxy not in self.blacklist:
                    return proxy
            raise RuntimeError('All proxies blacklisted')

    def report_failure(self, proxy):
        with self.lock:
            self.failures[proxy] = self.failures.get(proxy, 0) + 1
            if self.failures[proxy] >= self.max_failures:
                self.blacklist[proxy] = time.time()
                self.failures[proxy] = 0

    def report_success(self, proxy):
        with self.lock:
            self.failures.pop(proxy, None)

def fetch_with_retry(url, pool, max_retries=5):
    for attempt in range(max_retries):
        proxy = pool.get()
        try:
            r = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=15
            )
            if r.status_code in (200, 304):
                pool.report_success(proxy)
                return r
            elif r.status_code in (429, 503):
                pool.report_failure(proxy)
                continue
        except (requests.Timeout, requests.ConnectionError):
            pool.report_failure(proxy)
    return None

aiohttp : rotation async pour un throughput élevé

Pour scraper plus de 10K URLs/minute, requests ne suffit pas. aiohttp offre un throughput asynchrone :

import asyncio
import aiohttp
import random

PROXIES = ['http://user:[email protected]:7777', ...]
SEMAPHORE = asyncio.Semaphore(50)  # 50 parallel connections

async def fetch(session, url):
    async with SEMAPHORE:
        proxy = random.choice(PROXIES)
        try:
            async with session.get(
                url,
                proxy=proxy,
                timeout=aiohttp.ClientTimeout(total=15)
            ) as r:
                if r.status == 200:
                    return await r.text()
                return None
        except (asyncio.TimeoutError, aiohttp.ClientError):
            return None

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        return await asyncio.gather(*tasks)

# Usage
urls = ['https://example.com/' + str(i) for i in range(1000)]
results = asyncio.run(fetch_all(urls))

Scrapy : une ProxyMiddleware avec rotation

Scrapy dispose d'un support proxy intégré, mais la rotation nécessite un middleware personnalisé :

# middlewares.py
import random

class RotatingProxyMiddleware:
    def __init__(self, proxies):
        self.proxies = proxies

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.getlist('PROXY_POOL'))

    def process_request(self, request, spider):
        request.meta['proxy'] = random.choice(self.proxies)

    def process_exception(self, request, exception, spider):
        # if a proxy dies, retry with another
        request.meta['proxy'] = random.choice(self.proxies)
        return request

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RotatingProxyMiddleware': 350,
}
PROXY_POOL = [
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
]
DOWNLOAD_TIMEOUT = 15
RETRY_TIMES = 3

Problèmes courants et solutions

• Erreurs SSL avec des proxys residential — ajoutez verify=False pour le débogage. En production, configurez votre propre bundle CA.
• HTTP 407 (Proxy Authentication) — vérifiez le format user:password. Certains fournisseurs exigent l'encodage URL des caractères spéciaux.
• Connection reset sur du residential — normal, les sessions proxy tombent. Implémentez une logique de retry.
• HTTP 429 depuis le site cible — un rate limit. Ajoutez un sleep et faites tourner les IP plus agressivement.
• "Tunnel connection failed" — le format user:pass est incorrect pour HTTPS. Essayez https:// dans l'URL du proxy.

Pour des projets sérieux, utilisez un paquet dédié comme requests-ip-rotator (AWS API Gateway en guise de proxy) ou le SDK entreprise de Bright Data.