How to Rotate Proxies in Python

Sending many requests from a single IP triggers rate limits, CAPTCHAs, and IP bans on target sites. A rotating proxy pool spreads requests across multiple IPs so each IP stays below the target site's detection threshold.

The rotation strategy you choose affects both performance and detection avoidance. Per-request rotation is the most common for stateless scraping; per-session is required when the target site tracks session consistency across requests (cookies, login state).

proxy pool

Round-robin assigns proxies in a fixed repeating sequence, proxy 1, proxy 2, proxy 3, proxy 1, …, ensuring every proxy gets equal use.

import itertools
import requests

PROXIES = [
    "http://your-proxy-1.sparkproxy.io:10000",
    "http://your-proxy-2.sparkproxy.io:10001",
    "http://your-proxy-3.sparkproxy.io:10002",
]

proxy_cycle = itertools.cycle(PROXIES)

def get(url: str) -> requests.Response:
    proxy = next(proxy_cycle)
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=10)

for url in ["https://httpbin.org/ip"] * 6:
    resp = get(url)
    print(resp.json()["origin"])

Random rotation picks a proxy uniformly at random for each request. It provides better entropy than round-robin when one proxy in the pool is degraded, a round-robin pool will keep hitting the bad proxy every N requests, while random rotation naturally reduces its frequency.

import random
import requests

PROXIES = [
    "http://your-proxy-1.sparkproxy.io:10000",
    "http://your-proxy-2.sparkproxy.io:10001",
    "http://your-proxy-3.sparkproxy.io:10002",
]

def get(url: str) -> requests.Response:
    proxy = random.choice(PROXIES)
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=10)

random.choice() is thread-safe in CPython, it uses the module-level Random instance whose __next__ method is protected by the GIL. However, random.seed() is not thread-safe; if you call it from multiple threads, use separate random.Random instances instead.

Use weighted rotation when proxies have different reliability, speed, or geographic value. random.choices() accepts a weights parameter:

import random
import requests

# (proxy_url, weight), higher weight = selected more often
PROXY_POOL = [
    ("http://us-fast-1.sparkproxy.io:10000", 5),
    ("http://us-fast-2.sparkproxy.io:10001", 5),
    ("http://eu-medium.sparkproxy.io:10002", 3),
    ("http://fallback.sparkproxy.io:10003",  1),
]

PROXIES   = [p for p, _ in PROXY_POOL]
WEIGHTS   = [w for _, w in PROXY_POOL]

def get(url: str) -> requests.Response:
    proxy = random.choices(PROXIES, weights=WEIGHTS, k=1)[0]
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=10)

random.choices() normalizes weights automatically, no need to sum to 1.0. To dynamically adjust weights based on success/failure rates, update the weights list based on response metrics (latency, error count) between batches.

Python proxy rotation without retry logic fails silently when a proxy goes down, you get ProxyError or ConnectionError and lose the request. Use urllib3.Retry with a requests.HTTPAdapter to automatically retry with the next attempt:

import random
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

PROXIES = [
    "http://your-proxy-1.sparkproxy.io:10000",
    "http://your-proxy-2.sparkproxy.io:10001",
    "http://your-proxy-3.sparkproxy.io:10002",
]

def make_session(proxy: str) -> requests.Session:
    """Create a session bound to one proxy with retry logic."""
    retry = Retry(
        total=3,
        backoff_factor=0.5,           # 0s, 0.5s, 1s between retries
        status_forcelist=[407, 429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session = requests.Session()
    session.mount("http://",  adapter)
    session.mount("https://", adapter)
    session.proxies = {"http": proxy, "https": proxy}
    return session


def get_with_rotation(url: str, retries: int = 3) -> requests.Response:
    """Try up to `retries` different proxies on failure."""
    proxies = random.sample(PROXIES, min(retries, len(PROXIES)))
    last_exc = None

    for proxy in proxies:
        session = make_session(proxy)
        try:
            resp = session.get(url, timeout=15)
            resp.raise_for_status()
            return resp
        except requests.exceptions.RequestException as exc:
            last_exc = exc
            continue  # Try next proxy

    raise RuntimeError(f"All {len(proxies)} proxies failed for {url}") from last_exc

Key parameters in Retry:

Note: 407 Proxy Authentication Required in status_forcelist handles the case where proxy credentials expire mid-session. The retry will re-attempt the full request.

For multi-threaded scrapers, use a queue.Queue-backed pool. queue.Queue.get() and queue.Queue.put() are both thread-safe by design. The pool marks failed proxies and re-queues healthy ones:

import queue
import threading
import time
import requests


class ProxyPool:
    """Thread-safe rotating proxy pool with failure tracking."""

    def __init__(self, proxies: list[str], max_failures: int = 3):
        self._pool: queue.Queue[str] = queue.Queue()
        self._failures: dict[str, int] = {}
        self._lock = threading.Lock()
        self._max_failures = max_failures

        for proxy in proxies:
            self._pool.put(proxy)
            self._failures[proxy] = 0

    def get(self, block: bool = True, timeout: float = 5.0) -> str:
        """Acquire a proxy from the pool (blocks if empty)."""
        return self._pool.get(block=block, timeout=timeout)

    def release(self, proxy: str, success: bool) -> None:
        """Return proxy to pool. Remove it if it exceeded failure threshold."""
        with self._lock:
            if success:
                self._failures[proxy] = 0
                self._pool.put(proxy)
            else:
                self._failures[proxy] += 1
                if self._failures[proxy] < self._max_failures:
                    self._pool.put(proxy)
                # else: proxy is evicted from the pool

    def size(self) -> int:
        return self._pool.qsize()


# Usage
pool = ProxyPool([
    "http://your-proxy-1.sparkproxy.io:10000",
    "http://your-proxy-2.sparkproxy.io:10001",
    "http://your-proxy-3.sparkproxy.io:10002",
])

def fetch(url: str) -> str | None:
    proxy = pool.get(timeout=5.0)
    try:
        resp = requests.get(
            url,
            proxies={"http": proxy, "https": proxy},
            timeout=10,
        )
        resp.raise_for_status()
        pool.release(proxy, success=True)
        return resp.text
    except requests.exceptions.RequestException:
        pool.release(proxy, success=False)
        return None


# Multi-threaded example
from concurrent.futures import ThreadPoolExecutor

urls = [f"https://httpbin.org/ip?n={i}" for i in range(20)]

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch, urls))

print(f"Success: {sum(r is not None for r in results)}/{len(urls)}")

This pattern guarantees:

• No two threads use the same proxy concurrently (queue get() is atomic)
• Failed proxies are retired after max_failures consecutive failures
• Pool size is always known via size()

Automatic proxy rotation in async Python requires selecting a new proxy per request without blocking the event loop. Both httpx and aiohttp support per-request proxy configuration.

Scrapy uses a downloader middleware to assign a proxy to each Request before it goes out. The request.meta["proxy"] key is read by Scrapy's built-in HttpProxyMiddleware.

Custom rotation middleware (middlewares.py):

import random

class RotatingProxyMiddleware:
    PROXIES = [
        "http://your-proxy-1.sparkproxy.io:10000",
        "http://your-proxy-2.sparkproxy.io:10001",
        "http://your-proxy-3.sparkproxy.io:10002",
    ]

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

settings.py:

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.RotatingProxyMiddleware": 100,
    "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 110,
}

The middleware priority number controls execution order. Lower numbers run first, the custom middleware runs before Scrapy's built-in HttpProxyMiddleware, which reads request.meta["proxy"] and applies it.

For per-domain rotation (different proxies for different sites):

class RotatingProxyMiddleware:
    PROXY_MAP = {
        "us.example.com": [
            "http://us-1.sparkproxy.io:10000",
            "http://us-2.sparkproxy.io:10001",
        ],
        "eu.example.com": [
            "http://eu-1.sparkproxy.io:10002",
        ],
    }
    DEFAULT = ["http://fallback.sparkproxy.io:10003"]

    def process_request(self, request, spider):
        domain = request.url.split("/")[2]
        pool   = self.PROXY_MAP.get(domain, self.DEFAULT)
        request.meta["proxy"] = random.choice(pool)

Adding a proxy to your rotation pool without validating it first causes silent failures. Run a health check before every pool refresh:

import concurrent.futures
import requests

def check_proxy(proxy_url: str, test_url: str = "https://httpbin.org/ip", timeout: int = 8) -> dict:
    try:
        resp = requests.get(
            test_url,
            proxies={"http": proxy_url, "https": proxy_url},
            timeout=timeout,
        )
        resp.raise_for_status()
        return {
            "proxy":  proxy_url,
            "status": "ok",
            "exit_ip": resp.json().get("origin"),
            "latency_ms": int(resp.elapsed.total_seconds() * 1000),
        }
    except requests.exceptions.RequestException as exc:
        return {"proxy": proxy_url, "status": "failed", "error": str(exc)}


def filter_healthy(proxies: list[str], workers: int = 10) -> list[str]:
    """Return only working proxies from the list, tested in parallel."""
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        results = list(executor.map(check_proxy, proxies))

    healthy = [r["proxy"] for r in results if r["status"] == "ok"]
    print(f"{len(healthy)}/{len(proxies)} proxies healthy")
    return healthy


# Before building your pool, filter to healthy proxies only
RAW_PROXIES = [
    "http://your-proxy-1.sparkproxy.io:10000",
    "http://your-proxy-2.sparkproxy.io:10001",
    "http://bad-proxy.example.com:9999",
]

ACTIVE_PROXIES = filter_healthy(RAW_PROXIES)

Run filter_healthy() on pool initialization and periodically (every 5, 10 minutes for long-running scrapers) to evict dead proxies and re-admit recovered ones.