Proxy-Rotation in Python einrichten: requests, aiohttp, scrapy

Funktionierende Python-Codebeispiele für Proxy-Rotation in den gängigen Bibliotheken. Eigene Retry-Logik, Verwaltung des Proxy-Pools, Fehlerbehandlung, rate limiting.

requests: einfache Rotation

Die einfachste Option für On-Demand-Skripte. Übergib den Proxy direkt an den Aufruf:

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

Für residential mit sticky Session übergibst du das Session-Token im Benutzernamen:

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: ein Proxy-Pool mit Retry und Blacklist

Eine produktionsreife Klasse, die ausgefallene Proxys behandelt:

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: async Rotation für hohen throughput

Für scraping von 10K+ URLs/Minute reicht requests nicht aus. aiohttp liefert asynchronen throughput:

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: eine ProxyMiddleware mit Rotation

Scrapy hat eingebauten Proxy-Support, aber für Rotation braucht es eine eigene Middleware:

# 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

Häufige Probleme und Lösungen

• SSL-Fehler mit residential Proxys — füge zum Debuggen verify=False hinzu. In Produktion konfigurierst du dein eigenes CA-Bundle.
• HTTP 407 (Proxy Authentication) — prüfe das user:password-Format. Manche Anbieter verlangen URL-Encoding von Sonderzeichen.
• Connection reset bei residential — normal, Proxy-Sessions brechen ab. Implementiere eine Retry-Logik.
• HTTP 429 von der Zielseite — ein rate limit. Füge sleep ein und rotiere die IPs aggressiver.
• "Tunnel connection failed" — das user:pass-Format ist für HTTPS falsch. Versuche https:// in der Proxy-URL.

Für ernsthafte Projekte nutze ein dediziertes Paket wie requests-ip-rotator (AWS API Gateway als Proxy) oder das Enterprise-SDK von Bright Data.