Rotating proxies is easy. Surviving bans, timeouts and dead proxies is not. An 80-line Python rotator with health tracking and failover — copy-paste ready
TL;DR: Rotating a list of proxies is easy. Rotating them without your scraper dying when proxies get banned, time out, or return garbage is the part people get wrong. Below is a small, dependency-light proxy rotator that does retries, health tracking, and failover — copy-paste ready.
If you scrape anything at scale, you already know the pain: you paste a list of proxies into a loop, it works for ten minutes, then half of them get rate-limited, three are dead, one returns a captcha page with HTTP 200, and your job silently produces junk. A naive random.choice(proxies) doesn't cut it.
This post builds a rotator that treats proxies like what they are — an unreliable resource pool — and keeps your scraper running anyway.
What a real rotator needs
Before code, the requirements that separate a toy from something you'd run in production:
- Rotation — spread requests across the pool so no single IP gets hammered.
- Retries — a failed request should retry on a different proxy, not the same dead one.
- Health tracking — proxies that fail repeatedly get benched; proxies that recover come back.
- Failure classification — a timeout, a 407, a 429, and a captcha page are different problems and shouldn't be treated the same.
-
No heavy dependencies — just
requests. (An async version at the end.)
Step 1: model a proxy with health
Each proxy is more than a string — it carries a score we adjust as it succeeds or fails.
import time
from dataclasses import dataclass, field
@dataclass
class Proxy:
url: str # "http://user:pass@host:port"
fails: int = 0 # consecutive failures
disabled_until: float = 0.0 # unix time; 0 means active
def is_available(self) -> bool:
return time.time() >= self.disabled_until
def penalize(self, cooldown: float = 60.0):
self.fails += 1
# exponential backoff: 60s, 120s, 240s...
self.disabled_until = time.time() + cooldown * (2 ** (self.fails - 1))
def reward(self):
self.fails = 0
self.disabled_until = 0.0
Note the exponential backoff: a proxy that keeps failing gets benched for longer each time, instead of being retried into the ground.
Step 2: the rotator
import itertools
import threading
class ProxyRotator:
def __init__(self, proxy_urls):
self._proxies = [Proxy(u) for u in proxy_urls]
self._cycle = itertools.cycle(self._proxies)
self._lock = threading.Lock()
def get(self) -> Proxy:
"""Return the next available proxy, or wait for the soonest one."""
with self._lock:
for _ in range(len(self._proxies)):
p = next(self._cycle)
if p.is_available():
return p
# all benched — return the one that frees up soonest
return min(self._proxies, key=lambda p: p.disabled_until)
def alive_count(self) -> int:
return sum(p.is_available() for p in self._proxies)
The itertools.cycle gives round-robin rotation; the lock makes it thread-safe so you can use it from a thread pool.
Step 3: the request wrapper with retries and classification
This is the part that actually saves your job. We classify what went wrong and react accordingly.
import requests
# treat these as "the proxy or target is refusing us"
BAD_STATUS = {403, 407, 429, 502, 503}
def fetch(rotator: ProxyRotator, url: str, max_retries: int = 4, timeout: int = 20):
last_err = None
for attempt in range(max_retries):
proxy = rotator.get()
try:
r = requests.get(
url,
proxies={"http": proxy.url, "https": proxy.url},
timeout=timeout,
)
if r.status_code in BAD_STATUS:
proxy.penalize()
last_err = f"HTTP {r.status_code}"
continue
# crude captcha / block-page check on a 200
if "captcha" in r.text[:2000].lower():
proxy.penalize(cooldown=120)
last_err = "captcha page"
continue
proxy.reward()
return r
except (requests.Timeout, requests.ConnectionError) as e:
proxy.penalize()
last_err = type(e).__name__
continue
raise RuntimeError(f"all {max_retries} attempts failed, last: {last_err}")
Key details most tutorials miss:
- A 200 with a captcha is a failure. Checking status code alone is not enough — sites return 200 on block pages constantly.
- A 407 means proxy auth failed — penalize the proxy, don't blame the target.
- We retry on a fresh proxy each attempt, because retrying the same dead one is pointless.
Step 4: use it
PROXIES = [
"http://user:pass@p1.provider.com:8000",
"http://user:pass@p2.provider.com:8000",
"http://user:pass@p3.provider.com:8000",
]
rotator = ProxyRotator(PROXIES)
for target in ["https://httpbin.org/ip"] * 20:
try:
resp = fetch(rotator, target)
print(resp.json(), f"| alive: {rotator.alive_count()}")
except RuntimeError as e:
print("giving up on this URL:", e)
Run it and watch alive shrink and recover as proxies get benched and come back. That visibility alone will teach you more about your proxy pool's real quality than any provider's marketing page.
Async version (for real throughput)
requests is synchronous. For thousands of URLs, here's the aiohttp core — same rotator, async fetch:
import aiohttp
import asyncio
async def fetch_async(rotator, session, url, max_retries=4):
for _ in range(max_retries):
proxy = rotator.get()
try:
async with session.get(url, proxy=proxy.url, timeout=20) as r:
if r.status in BAD_STATUS:
proxy.penalize()
continue
proxy.reward()
return await r.text()
except (aiohttp.ClientError, asyncio.TimeoutError):
proxy.penalize()
return None
async def main(urls):
rotator = ProxyRotator(PROXIES)
async with aiohttp.ClientSession() as session:
tasks = [fetch_async(rotator, session, u) for u in urls]
return await asyncio.gather(*tasks)
(Note: aiohttp proxy auth needs the credentials in the proxy URL, which our format already has.)
What this won't fix
Honest limits, because your rotator is only as good as the proxies feeding it:
- A rotator can't fix a bad pool. If your provider hands you already-flagged datacenter IPs, no amount of retry logic saves you. Rotating dead weight is still dead weight.
- Rotating too aggressively can cause bans for stateful tasks (logins, carts). For account work you want sticky sessions, not rotation. Rotation is for stateless scraping.
- Free proxy lists will make this code look broken because 90% of the pool is dead. The logic is fine; the input is garbage.
For the pool itself, you want residential or mobile IPs with real session control. I run mine through ProxyUniverse — residential, mobile and ISP proxies that come as standard host:port:user:pass, so they drop straight into the PROXIES list above with no changes. After the 2026 provider shutdowns, I keep the pool spread across networks so one dying provider doesn't empty my rotator.
Wrap-up
The difference between a scraper that runs for ten minutes and one that runs for ten days is almost never the parsing code — it's how you handle the proxy layer failing. Health tracking + classified retries + failover is maybe 60 lines, and it's the 60 lines that matter.
Grab the code, drop in your own proxies, and watch the alive counter. If you build something on top of this (proxy scoring, per-domain pools, Prometheus metrics), I'd genuinely like to see it.
Question for the comments: how do you detect soft blocks — captcha pages and junk HTML that come back with a clean HTTP 200? That's the failure class this rotator handles worst, and I'm curious what checks people run in production. 👇
Top comments (0)