How to rotate proxies in Python (requests, aiohttp, Scrapy)

Proxy rotation in Python means giving each request a different exit IP. You either cycle a list of proxy URLs in your own code — itertools.cycle or random.choice plus health checks — or send every request to a rotating gateway that swaps the IP server-side. The list is cheap but stateful; the gateway removes the state.

The two models, side by side

Every tutorial you will find shows the same first ten lines: build a list, pick an element, pass a proxies dict to requests.get(). That part is trivial. What separates a working scraper from a script that dies at 3 a.m. is who owns the state — which IPs are burned, which are cooling down, which country a given request needs.

Most production systems end up using two of these: a gateway or API for hard targets, and a plain list for internal or tolerant endpoints. The code below covers both paths.

Rotate proxies in Python with requests

requests takes a per-scheme dictionary. The same URL usually works for both keys, because the client opens a CONNECT tunnel for HTTPS through the HTTP proxy. Note that requests also honours the HTTP_PROXY and HTTPS_PROXY environment variables, so a stray shell export can silently override your dict during debugging (requests docs).

import itertools, requests

with open('proxies.txt') as f:
    pool = [line.strip() for line in f if line.strip()]

rotator = itertools.cycle(pool)   # round robin

for url in urls:
    proxy = next(rotator)
    proxies = {'http': f'http://{proxy}', 'https': f'http://{proxy}'}
    try:
        r = requests.get(url, proxies=proxies, timeout=15)
        print(r.status_code, len(r.content), proxy)
    except requests.RequestException as err:
        print('failed', proxy, err)

One trap: if you set session.proxies on a requests.Session, every request in that session uses the same exit. Per-request rotation means passing proxies= to each call, or keeping one session per proxy so connection pooling and cookies stay aligned with the IP. Mixing a single cookie jar across many exits is a strong bot signal.

Keep the pool in JSON, not a text file

A flat ip:port list has no room for metadata, and you will need metadata the moment one target requires German IPs and another needs mobile. JSON scales better:

[
  {"url": "http://user:[email protected]:8000", "country": "de", "type": "residential"},
  {"url": "http://user:[email protected]:8001", "country": "us", "type": "datacenter"},
  {"url": "socks5://user:[email protected]:1080", "country": "gb", "type": "mobile"}
]

import json, random

pool = json.load(open('proxies.json'))
de_pool = [p['url'] for p in pool if p['country'] == 'de']
proxy = random.choice(de_pool)

Round robin spreads load evenly, which matters when your pool is small; random picking avoids the predictable request order that makes a pool easy to fingerprint as a group. In practice, random selection weighted by recent success rate beats both.

Health checks, retries and cooldowns

This is the part that decides whether rotation works. Two rules keep a pool alive: never delete an IP on a single failure, and never treat all failures the same. A connection reset is a proxy problem. A 403 with a challenge page is a target problem — the IP may be fine tomorrow. A 407 is your own credentials.

import random, time, requests

class ProxyPool:
    def __init__(self, proxies, cooldown=300, max_fails=3):
        self.state = {p: {'fails': 0, 'until': 0.0} for p in proxies}
        self.cooldown, self.max_fails = cooldown, max_fails

    def get(self):
        now = time.time()
        live = [p for p, s in self.state.items() if s['until'] < now]
        if not live:
            raise RuntimeError('every proxy is cooling down')
        return random.choice(live)

    def report(self, proxy, ok):
        s = self.state[proxy]
        if ok:
            s['fails'] = 0
            return
        s['fails'] += 1
        if s['fails'] >= self.max_fails:
            s['until'] = time.time() + self.cooldown
            s['fails'] = 0

BAD = {403, 407, 429, 502, 503}

def fetch(pool, url, attempts=4):
    for _ in range(attempts):
        proxy = pool.get()
        try:
            r = requests.get(url, proxies={'http': proxy, 'https': proxy}, timeout=15)
        except requests.RequestException:
            pool.report(proxy, False)
            continue
        if r.status_code in BAD:
            pool.report(proxy, False)
            continue
        pool.report(proxy, True)
        return r
    raise RuntimeError(f'all attempts failed: {url}')

Pre-flight validation against an IP echo endpoint is useful once, at startup, but it does not tell you whether the exit is blocked by your target. The only honest health signal is the response your target returns, which is why scoring belongs in the request path rather than in a separate checker script.

Async rotation with aiohttp

Sequential rotation wastes the pool: one proxy works while the rest idle. aiohttp accepts a proxy= string per request, so a semaphore plus a cycled rotator gives you concurrent IP rotation for scraping in about twenty lines (aiohttp docs).

import asyncio, itertools, aiohttp

async def fetch(session, sem, url, rotator):
    proxy = next(rotator)
    async with sem:
        try:
            async with session.get(url, proxy=proxy,
                                   timeout=aiohttp.ClientTimeout(total=20)) as r:
                body = await r.read()
                return url, r.status, len(body)
        except Exception as err:
            return url, None, repr(err)

async def main(urls, proxies, concurrency=20):
    rotator = itertools.cycle(proxies)
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, sem, u, rotator) for u in urls]
        for coro in asyncio.as_completed(tasks):
            print(await coro)

asyncio.run(main(urls, proxies))

Two things to add before you run this at volume: a per-domain rate limiter, because 20 concurrent requests from 20 different IPs to one small site is still 20 concurrent requests, and the same scoring logic from the synchronous example. Concurrency without cooldowns just burns the pool faster. If you are crawling rather than fetching a known list, the same pattern applies to a frontier queue — see our walkthrough on crawling in Python.

Scrapy and the rotating proxy Python ecosystem on GitHub

In Scrapy you do not write a rotator; you enable middleware. scrapy-rotating-proxies ships a rotating middleware plus a ban-detection middleware, tracks per-proxy state and re-checks dead proxies for you (PyPI).

DOWNLOADER_MIDDLEWARES = {
    'rotating_proxies.middlewares.RotatingProxyMiddleware': 610,
    'rotating_proxies.middlewares.BanDetectionMiddleware': 620,
}
ROTATING_PROXY_LIST_PATH = 'proxies.txt'
RETRY_TIMES = 5
RETRY_HTTP_CODES = [408, 429, 500, 502, 503, 504]

Beyond that, the GitHub landscape splits into three kinds of package. Reference repos, like the sync and async example scripts in oxylabs/Rotating-Proxies-With-Python, are worth reading once and then rewriting for your own error taxonomy. Free-proxy harvesters such as Swiftshadow fetch and cache public proxy lists so your code can ask for a working one; they are fine for a demo, and unusable for anything with a deadline, because public proxies are shared, logged by unknown parties and often already blocked. Third, provider SDKs, which are really thin wrappers over a gateway URL.

When you evaluate any Python proxy library, ask four questions: does it persist pool state across restarts, does it distinguish transport errors from target-side bans, can it target a country per request, and does it let you inject your own success predicate? Most fail the last two.

Gateway rotation: delete the pool code

The alternative is to stop rotating in Python at all. A rotating proxy endpoint gives you a fresh IP per request behind one hostname, so your code collapses to a single static proxies dict and your retry loop only handles transport errors. Country selection and session persistence move into the username string rather than your pool metadata.

import requests

proxy = 'http://USER-country-de-session-a91f:[email protected]:7000'
r = requests.get('https://example.com',
                 proxies={'http': proxy, 'https': proxy}, timeout=20)

Use rotating exits for stateless fetches and a sticky session — up to 120 minutes on QuanticData's rotating network — when a flow needs the same IP across login, pagination or cart steps. Pick the network by target tolerance: residential IPs from $0.80/GB for consumer sites that score ASNs, datacenter from $0.50/GB for APIs and tolerant endpoints where speed dominates.

What rotation actually costs

Proxy bandwidth is billed on bytes transferred, including the bytes of every block page you get served. The numbers below are illustrative: they assume 100–500 KB of HTML for a text-only fetch and 2–5 MB once a headless browser pulls scripts, fonts and images. Your mileage depends entirely on page weight.

The honest reading: raw proxies win on unit price for lightweight, high-volume, low-block fetching, especially with datacenter or IPv6 exits. A per-success API wins when block rates are non-trivial, when pages need rendering, or when the bytes you would burn on retries and challenge pages exceed the price of the successful page. Measure your own block rate for a week before choosing — that single number decides it.

Rotation as a tool call

For agent workloads the calculus changes again. An LLM should not be maintaining a cooldown table. It should ask for a page and get clean Markdown or JSON back. That is one POST against the scraping API, with residential rotation underneath:

import os, requests

res = requests.post(
    'https://api.quanticdata.io/v1/scrape',
    headers={'Authorization': f"Bearer {os.environ['QD_API_KEY']}"},
    json={'url': 'https://example.com'},
    timeout=60,
)
payload = res.json()
print(payload['success'], payload['usage']['cost_usd'])
print(payload['data']['markdown'][:200])

curl https://api.quanticdata.io/v1/scrape \
  -H "Authorization: Bearer $QD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com"}'

Every endpoint answers with the same envelope — success, data, usage — and calls that fought through blocks include a per-attempt retry log, so slow responses have an explanation in the response itself. For known URL lists, POST /v1/batch takes up to 1,000 URLs and you poll GET /v1/batch/{jobId}; async jobs auto-refund the unfetched share. The same tools are exposed to Claude, Cursor and Cline through the MCP server, which is the shortest path from "my agent needs the web" to working calls.

Checklist before you scale

1. Classify failures into three buckets — transport, target ban, credentials — and score only the first two against the proxy.
2. Give every exit a cooldown instead of deleting it; persist the state so a restart does not reset your knowledge.
3. Bind one cookie jar, one User-Agent and one TLS session per exit IP. Rotating IPs while sharing identity is worse than not rotating.
4. Rate-limit per target domain independently of pool size.
5. Log the exit IP with every response so you can reconstruct which subnet got blocked and when.
6. Track cost per successful page, not cost per GB. It is the only figure that compares a self-managed pool against an API.

Rotation is not the hard part of scraping — deciding what counts as a failure, and what to do about it, is. Get that loop right and the rest is ten lines of requests.