Configuration des proxies dans Python requests et aiohttp : guide complet avec des exemples de code

Lors du développement de parseurs, de l'automatisation de la collecte de données ou des tests de services web, il est souvent nécessaire d'utiliser des serveurs proxy depuis Python. Les bibliothèques requests et aiohttp fournissent des mécanismes flexibles pour travailler avec des proxies, mais leur configuration présente des nuances importantes. Dans ce guide, nous examinerons les approches synchrones et asynchrones, fournirons des exemples pour les proxies HTTP et SOCKS5, et aborderons la rotation IP et la gestion des erreurs.

Configuration de base des proxies dans requests

La bibliothèque requests est la norme pour les requêtes HTTP en Python. La configuration des proxies se fait via le paramètre proxies, qui accepte un dictionnaire avec les protocoles et les adresses des serveurs proxy.

Exemple simple avec un proxy HTTP :

import requests

# Configuration du proxy
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Exécution de la requête via le proxy
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Notez que pour les requêtes HTTPS, le protocole http:// est également spécifié dans la valeur du proxy (et non https://). Cela est dû au fait que la connexion au serveur proxy est établie via HTTP, puis un tunnel est créé pour le trafic HTTPS via la méthode CONNECT.

Utilisation des variables d'environnement :

La bibliothèque requests lit automatiquement les proxies à partir des variables d'environnement HTTP_PROXY et HTTPS_PROXY :

import os
import requests

# Configuration via les variables d'environnement
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Le proxy sera appliqué automatiquement
response = requests.get('https://httpbin.org/ip')
print(response.json())

Cette approche est pratique pour la conteneurisation (Docker) ou lorsque les proxies sont configurés au niveau du système. Cependant, pour plus de flexibilité, il est recommandé de transmettre explicitement le paramètre proxies.

Authentification et SOCKS5 dans requests

La plupart des services de proxy commerciaux nécessitent une authentification par nom d'utilisateur et mot de passe. Dans requests, cela se fait via un format d'URL avec les informations d'identification.

Proxy HTTP avec authentification :

import requests

# Format : 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())

Configuration d'un proxy SOCKS5 :

Pour travailler avec SOCKS5, une bibliothèque supplémentaire requests[socks] ou PySocks est requise. Installation :

pip install requests[socks]

Exemple d'utilisation de SOCKS5 :

import requests

# SOCKS5 sans authentification
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 avec authentification
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())

Les proxies SOCKS5 sont particulièrement utiles lors de l'utilisation de proxies résidentiels, car ce protocole permet un tunneling de trafic plus fiable et prend en charge UDP (nécessaire pour certaines applications).

Rotation des proxies dans requests

Lors du parsing de grandes quantités de données, l'utilisation d'une seule adresse IP peut entraîner des blocages. La rotation des proxies consiste à changer cycliquement d'IP pour répartir la charge et contourner les limites de taux.

Rotation simple via une liste :

import requests
import itertools

# Liste des serveurs proxy
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# Création d'un itérateur infini
proxy_pool = itertools.cycle(proxy_list)

# Exécution des requêtes avec rotation
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"Requête {i+1} : IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Erreur avec le proxy {proxy} : {e}")

Rotation avec des sessions pour conserver les 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)

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

Rotation aléatoire pour l'imprévisibilité :

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}

# Chaque requête avec un proxy aléatoire
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Requête {i+1} : {response.json()['origin']}")

La rotation aléatoire est plus efficace lors de l'utilisation de sites qui suivent les modèles de requêtes. Le changement séquentiel d'IP peut sembler suspect, tandis que le choix aléatoire imite le comportement de différents utilisateurs.

Configuration des proxies dans aiohttp

La bibliothèque aiohttp est conçue pour des requêtes HTTP asynchrones et est critique pour les parseurs à forte charge. La configuration des proxies diffère de requests — le paramètre proxy (au singulier) est utilisé.

Exemple de base avec un 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)

# Lancement
asyncio.run(fetch_with_proxy())

Proxy avec authentification :

Dans aiohttp, l'authentification est transmise via l'objet aiohttp.BasicAuth ou directement dans l'URL :

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Variante 1 : Informations d'identification dans l'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())

# Variante 2 : Via BasicAuth (pour certains 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 dans aiohttp :

Pour SOCKS5, la bibliothèque aiohttp-socks est requise :

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

Lors de l'utilisation de proxies mobiles pour le parsing des réseaux sociaux ou des marketplaces, il est recommandé d'utiliser aiohttp — l'asynchronicité permet de traiter des centaines de requêtes en parallèle sans bloquer le fil d'exécution.

Rotation asynchrone et pool de proxies

Pour les parseurs à forte charge, une rotation efficace des proxies avec gestion des pannes et remplacement automatique des IP non fonctionnelles est critique. Examinons des modèles avancés pour aiohttp.

Classe pour gérer 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]:
        """Obtenir le prochain proxy fonctionnel"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Tous les proxies sont indisponibles
    
    def mark_failed(self, proxy: str):
        """Marquer le proxy comme non fonctionnel"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} marqué comme indisponible")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Exécuter une requête avec changement automatique de proxy en cas d'erreur"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Tous les proxies sont indisponibles")
            
            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"Erreur avec le proxy {proxy} : {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Échec de la requête après {max_retries} tentatives")

# Utilisation
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:
        # Exécution de 10 requêtes avec rotation automatique
        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"Requête {i+1} échouée : {result}")
            else:
                print(f"Requête {i+1} : IP = {result.get('origin')}")

asyncio.run(main())

Traitement parallèle avec limitation de la concurrence :

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Limitation des requêtes simultanées
        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 requêtes
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Limitation : pas plus de 10 requêtes simultanées
    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)
        
        # Analyse des résultats
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Requêtes réussies : {len(successful)}")
        print(f"Requêtes échouées : {len(failed)}")

asyncio.run(main())

L'utilisation de asyncio.Semaphore est critique lors de l'utilisation de proxies — un trop grand nombre de connexions simultanées via une seule IP peut entraîner un blocage de la part du site cible ou du fournisseur de proxy.

Gestion des erreurs et des délais d'attente

Travailler avec des proxies est associé à un nombre accru d'erreurs : délais d'attente, ruptures de connexion, échecs des serveurs proxy. Une bonne gestion des erreurs est la clé de la stabilité du parseur.

Erreurs typiques lors de l'utilisation de proxies :

Gestion des erreurs dans 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),  # (délai d'attente de connexion, délai d'attente de lecture)
                allow_redirects=True,
                verify=True  # Vérification du certificat SSL
            )
            response.raise_for_status()  # Déclenche une exception pour les erreurs 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Tentative {attempt + 1} : Proxy indisponible - {e}")
        except ConnectTimeout as e:
            print(f"Tentative {attempt + 1} : Délai d'attente de connexion - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Tentative {attempt + 1} : Erreur HTTP {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Erreur d'authentification du proxy !")
                break  # Ne pas réessayer en cas d'erreur d'autorisation
        except RequestException as e:
            print(f"Tentative {attempt + 1} : Erreur générale - {e}")
        
        if attempt < max_retries - 1:
            print(f"Réessayer dans 2 secondes...")
            import time
            time.sleep(2)
    
    raise Exception(f"Échec de la requête après {max_retries} tentatives")

# Utilisation
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"Erreur critique : {e}")

Gestion des erreurs dans 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"Tentative {attempt + 1} : Erreur de connexion au proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Tentative {attempt + 1} : Délai d'attente")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Tentative {attempt + 1} : Erreur HTTP du proxy - {e}")
            if e.status == 407:
                print("Erreur d'authentification du proxy !")
                break
        except ClientError as e:
            print(f"Tentative {attempt + 1} : Erreur générale du client - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Échec de la requête après {max_retries} tentatives")

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"Erreur critique : {e}")

asyncio.run(main())

Configuration des délais d'attente :

Une bonne configuration des délais d'attente est cruciale pour la stabilité. Valeurs recommandées :

• Délai d'attente de connexion : 5-10 secondes (temps pour établir une connexion avec le proxy)
• Délai d'attente de lecture : 10-30 secondes (temps pour recevoir une réponse du site cible)
• Délai d'attente total : 30-60 secondes (temps total de la requête)

Pour les proxies résidentiels lents, il est recommandé d'augmenter les délais d'attente à 20-30 secondes par connexion, car le routage via de véritables fournisseurs peut prendre plus de temps.

Meilleures pratiques et optimisation

Travailler efficacement avec des proxies nécessite de suivre un ensemble de règles pour minimiser les blocages et maximiser les performances.

1. Utiliser Session pour réutiliser les connexions :

# requests : Session réutilise les connexions TCP
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Plus rapide que requests.get()

# aiohttp : Session est obligatoire pour l'asynchronicité
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Définir des User-Agent et des en-têtes réalistes :

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': 'fr-FR,fr;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. Limiter le taux de requêtes (requêtes par seconde) :

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

# Utilisation : pas plus de 2 requêtes par seconde
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

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

4. Journaliser et surveiller les 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
        
        # Journaliser tous les 10 requêtes
        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} requêtes, succès {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# Dans le code de requête
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"Erreur avec le proxy {proxy} : {e}")

5. Mise en cache DNS pour accélérer :

# aiohttp avec mise en cache 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:
    # Les requêtes utiliseront le cache DNS pendant 5 minutes
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Gestion des CAPTCHA et des blocages :

Comparaison entre requests et aiohttp pour les proxies

Le choix entre requests et aiohttp dépend de la tâche et du volume de données. Examinons les différences clés.

Quand utiliser requests :

• Scripts simples pour des tâches ponctuelles
• Prototypage et tests
• Petit volume de requêtes (jusqu'à 100 par minute)
• Lorsque la simplicité du code et la lisibilité sont importantes
• Intégration avec des bibliothèques synchrones

Quand utiliser aiohttp :

• Parsing de grandes quantités de données (milliers de pages)
• Surveillance de nombreuses sources en temps réel
• Services API à forte charge
• Lorsque la vitesse de traitement est critique
• Travail avec WebSocket via proxy

Comparaison pratique des performances :

# Test : 100 requêtes via proxy

# requests (synchronisé) - ~50 secondes
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} secondes")

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

Lors de l'utilisation de proxies de centres de données pour le parsing à grande vitesse, aiohttp montre un avantage de 10 à 20 fois par rapport à requests grâce au traitement parallèle des requêtes.

Conclusion

La configuration des proxies en Python via les bibliothèques requests et aiohttp est une compétence fondamentale pour le développement de parseurs, l'automatisation de la collecte de données et le contournement des restrictions géographiques. La bibliothèque requests est adaptée pour des scripts simples et le prototypage grâce à une API synchrone claire, tandis qu'aiohttp offre des performances élevées lors du traitement de milliers de requêtes via une architecture asynchrone.

Points clés pour un travail efficace avec des proxies en Python : gestion appropriée des erreurs et des délais d'attente, mise en œuvre de la rotation des adresses IP pour répartir la charge, utilisation de Session pour réutiliser les connexions, configuration d'en-têtes et de User-Agent réalistes, surveillance des performances des serveurs proxy. Pour les proxies SOCKS5, des bibliothèques supplémentaires sont nécessaires — requests[socks] ou aiohttp-socks.

Lors du choix du type de proxy pour le parsing, tenez compte de la spécificité de la tâche : pour les parseurs à forte charge avec des milliers de requêtes, optez pour des proxies de centres de données rapides, pour contourner des systèmes anti-bot stricts et travailler avec des réseaux sociaux, privilégiez les proxies résidentiels avec de véritables IP d'utilisateurs domestiques, et pour les tâches nécessitant un maximum d'anonymat et l'imitation du trafic mobile, les proxies mobiles avec des IP d'opérateurs de téléphonie mobile sont optimaux.

Si vous envisagez de développer des parseurs à haute performance ou d'automatiser la collecte de données à partir de nombreuses sources, nous vous recommandons d'essayer des proxies résidentiels — ils offrent un haut niveau d'anonymat, un risque minimal de blocages et un fonctionnement stable avec la plupart des services web sécurisés. Pour des tâches techniques à haute vitesse de traitement, les proxies de centres de données avec une faible latence et une grande capacité de bande passante conviennent également.