Setting up proxy rotation in Python: requests, aiohttp, scrapy

Working code samples for proxy rotation in popular Python libraries. Custom retry logic, proxy pool management, error handling, rate limiting.

requests: basic rotation

The simplest pattern for on-demand scripts. Pass the proxy into the call:

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

For residential with a sticky session — pass the session token in user:

import secrets
session_id = secrets.token_hex(8)
proxy = f'http://customer-XXX-session-{session_id}:[email protected]:7000'
# Now every request with this session token comes from the same IP

requests: a proxy pool with retry and blacklisting

A production-ready class with handling of failed proxies:

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()
            # Clean 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 for high throughput

For 10K+ URLs/minute requests isn't enough. aiohttp delivers async throughput:

import asyncio
import aiohttp
import random

PROXIES = ['http://user:[email protected]:7777', ...]
SEMAPHORE = asyncio.Semaphore(50)  # 50 concurrent 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: ProxyMiddleware with rotation

Scrapy has built-in proxy support, but rotation needs custom 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 died — 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

Common issues and fixes

• SSL errors with residential proxies — add verify=False for debug. In production set up your own CA bundle.
• HTTP 407 (Proxy Authentication) — verify the user:password format. Some providers require URL-encoding of special characters.
• Connection reset on residential — normal; proxy sessions die. Implement retry logic.
• HTTP 429 from the target — rate-limit. Add a sleep and rotate IPs more aggressively.
• "Tunnel connection failed" — wrong user:pass format for HTTPS. Try https:// in the proxy URL.

For serious projects use a dedicated package like requests-ip-rotator (for AWS API Gateway as proxy) or an enterprise SDK like Bright Data's.