Implementing a Proxy‑Aware Token Bucket Rate Limiter in Python for Distributed Scrapers

Why a Token Bucket for Proxy‑Aware Scrapers

When you run a distributed scraper behind a rotating proxy pool, each IP address has its own rate‑limit ceiling. A naïve time.sleep() between requests creates bursty traffic: a fast proxy may fire dozens of requests in a second while a slower one sits idle, quickly triggering bans. A token‑bucket algorithm solves this by granting each proxy a steady stream of "tokens" (request credits) that refill at a configurable rate. The scraper only sends a request when a token is available, guaranteeing smooth, predictable traffic per IP.

Key benefits:

• Per‑proxy fairness – every proxy obeys its own limit, preventing a single hot IP from exhausting the pool.
• Burst tolerance – buckets can hold a small surplus, allowing short spikes (e.g., handling a redirect chain) without immediate throttling.
• Composability – the limiter works with any async or sync HTTP client and plugs into existing proxy‑rotation logic.

Core Concepts

Token Bucket Algorithm

A token bucket has three parameters:

• capacity – maximum tokens the bucket can hold (burst allowance).
• refill_rate – tokens added per second (steady‑state throughput).
• tokens – current available tokens (float for sub‑second precision).

On each request the bucket attempts to consume(1). If tokens >= 1, the request proceeds and tokens -= 1. Otherwise the caller waits until enough tokens accumulate (wait_time = (1 - tokens) / refill_rate).

Mapping Buckets to Proxies

Each proxy in the pool gets its own bucket instance. When the rotation logic selects a proxy, the scraper also fetches the associated bucket. This design keeps the limiter stateless from the scraper’s perspective – it simply asks "can I use this proxy now?".

Implementation Overview

We’ll build four small, testable components:

1. ProxyPool – yields (proxy_url, proxy_id) pairs and reports health.
2. TokenBucket – core algorithm with async take() method.
3. BucketManager – creates, stores, and retrieves buckets keyed by proxy_id.
4. ScraperWorker – main loop that acquires a proxy, waits for its bucket, performs the request, and returns the proxy to the pool.

All code uses Python 3.11+, asyncio, and aiohttp for async HTTP. The same ideas translate to synchronous code or other languages.

Step‑by‑Step Implementation

1. Proxy Pool Abstraction

# proxy_pool.py
import random
from typing import AsyncIterator, Tuple

class ProxyPool:
    """Simple round‑robin pool with health tracking."""
    def __init__(self, proxies: list[str]):
        self._proxies = [{'url': p, 'id': i, 'healthy': True} for i, p in enumerate(proxies)]
        self._index = 0

    async def acquire(self) -> Tuple[str, int]:
        """Return next healthy proxy (url, id)."""
        for _ in range(len(self._proxies)):
            proxy = self._proxies[self._index]
            self._index = (self._index + 1) % len(self._proxies)
            if proxy['healthy']:
                return proxy['url'], proxy['id']
        raise RuntimeError('No healthy proxies available')

    def mark_unhealthy(self, proxy_id: int):
        for p in self._proxies:
            if p['id'] == proxy_id:
                p['healthy'] = False
                break

    def mark_healthy(self, proxy_id: int):
        for p in self._proxies:
            if p['id'] == proxy_id:
                p['healthy'] = True
                break

2. Token Bucket Class

# token_bucket.py
import asyncio
import time
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    capacity: float          # max tokens
    refill_rate: float       # tokens per second
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)

    def __post_init__(self):
        self._tokens = self.capacity
        self._last_refill = time.monotonic()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        new_tokens = elapsed * self.refill_rate
        self._tokens = min(self.capacity, self._tokens + new_tokens)
        self._last_refill = now

    async def take(self, tokens: float = 1.0) -> None:
        """Block until `tokens` are available, then consume them."""
        async with self._lock:
            while True:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return
                # wait just enough for the missing tokens
                deficit = tokens - self._tokens
                wait = deficit / self.refill_rate
                await asyncio.sleep(wait)

3. Bucket Manager

# bucket_manager.py
from token_bucket import TokenBucket

class BucketManager:
    """Holds a TokenBucket per proxy_id."""
    def __init__(self, capacity: float, refill_rate: float):
        self._capacity = capacity
        self._refill_rate = refill_rate
        self._buckets: dict[int, TokenBucket] = {}

    def get_bucket(self, proxy_id: int) -> TokenBucket:
        if proxy_id not in self._buckets:
            self._buckets[proxy_id] = TokenBucket(self._capacity, self._refill_rate)
        return self._buckets[proxy_id]

    def remove_bucket(self, proxy_id: int):
        self._buckets.pop(proxy_id, None)

4. Scraper Worker Loop

# scraper_worker.py
import asyncio
import aiohttp
from proxy_pool import ProxyPool
from bucket_manager import BucketManager

class ScraperWorker:
    def __init__(
        self,
        pool: ProxyPool,
        bucket_mgr: BucketManager,
        target_urls: list[str],
        concurrency: int = 10,
    ):
        self.pool = pool
        self.bucket_mgr = bucket_mgr
        self.target_urls = target_urls
        self.semaphore = asyncio.Semaphore(concurrency)
        self.session: aiohttp.ClientSession | None = None

    async def _fetch(self, url: str, proxy_url: str, proxy_id: int) -> dict:
        bucket = self.bucket_mgr.get_bucket(proxy_id)
        await bucket.take()                     # rate‑limit per proxy
        async with self.semaphore:
            try:
                async with self.session.get(url, proxy=proxy_url, timeout=10) as resp:
                    data = await resp.text()
                    return {'url': url, 'status': resp.status, 'len': len(data)}
            except Exception as exc:
                # on network error, mark proxy unhealthy and retry later
                self.pool.mark_unhealthy(proxy_id)
                raise

    async def run(self):
        self.session = aiohttp.ClientSession()
        tasks = []
        for url in self.target_urls:
            tasks.append(asyncio.create_task(self._process_url(url)))
        await asyncio.gather(*tasks, return_exceptions=True)
        await self.session.close()

    async def _process_url(self, url: str):
        while True:
            proxy_url, proxy_id = await self.pool.acquire()
            try:
                result = await self._fetch(url, proxy_url, proxy_id)
                print(f"[OK] {url} via proxy {proxy_id} -> {result['status']}")
                self.pool.mark_healthy(proxy_id)
                break
            except Exception:
                # simple retry with next proxy
                await asyncio.sleep(0.5)
                continue

5. Wiring It Together

# main.py
import asyncio
from proxy_pool import ProxyPool
from bucket_manager import BucketManager
from scraper_worker import ScraperWorker

PROXIES = [
    "http://user:pass@proxy1.example.com:8000",
    "http://user:pass@proxy2.example.com:8000",
    "http://user:pass@proxy3.example.com:8000",
]
TARGETS = [f"https://httpbin.org/get?page={i}" for i in range(100)]

async def main():
    pool = ProxyPool(PROXIES)
    # 5 req/s per proxy, burst of 10
    buckets = BucketManager(capacity=10, refill_rate=5)
    worker = ScraperWorker(pool, buckets, TARGETS, concurrency=20)
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

Handling Edge Cases

Situation	Strategy
Proxy becomes permanently banned	After N consecutive failures, call pool.mark_unhealthy(proxy_id) and bucket_mgr.remove_bucket(proxy_id) to free memory.
Dynamic proxy addition	BucketManager.get_bucket lazily creates a bucket, so new proxies work instantly.
Unequal proxy performance	Tune refill_rate per‑proxy by storing a custom rate in BucketManager (e.g., a dict proxy_id -> rate).
Clock drift / async delays	time.monotonic() is immune to system‑time changes; the lock inside TokenBucket.take serialises refills for a given proxy.

Testing the Limiter

A quick unit test verifies the bucket respects the configured rate.

# test_token_bucket.py
import asyncio
import pytest
from token_bucket import TokenBucket

@pytest.mark.asyncio
async def test_refill_rate():
    bucket = TokenBucket(capacity=5, refill_rate=10)  # 10 tokens/s
    start = asyncio.get_event_loop().time()
    for _ in range(5):
        await bucket.take()          # should be instant (burst)
    await bucket.take()              # 6th token -> wait ~0.1s
    elapsed = asyncio.get_event_loop().time() - start
    assert 0.09 < elapsed < 0.2      # allow scheduling jitter

Run with pytest -q.

Deploying in a Distributed Setting

When workers run on multiple machines, each process would otherwise maintain its own bucket state, breaking the per‑IP contract. Two practical approaches:

1. Centralised Redis buckets – store tokens and last_refill as a hash per proxy_id. Use a Lua script for atomic take (check‑and‑decrement). Workers call EVALSHA instead of local TokenBucket.take().
2. Proxy‑side rate limiting – many premium proxy providers (including RoProxy) enforce per‑IP limits upstream. In that case the local bucket acts only as a client‑side guard to avoid unnecessary 429 responses.

Example Redis Lua snippet (atomic take):

-- KEYS[1] = proxy_id, ARGV[1] = capacity, ARGV[2] = refill_rate, ARGV[3] = now_ms
local bucket = redis.call('HMGET', KEYS[1], 'tokens', 'last')
local tokens = tonumber(bucket[1]) or tonumber(ARGV[1])
local last = tonumber(bucket[2]) or tonumber(ARGV[3])
local elapsed = (tonumber(ARGV[3]) - last) / 1000
tokens = math.min(tonumber(ARGV[1]), tokens + elapsed * tonumber(ARGV[2]))
if tokens >= 1 then
    tokens = tokens - 1
    redis.call('HMSET', KEYS[1], 'tokens', tokens, 'last', ARGV[3])
    return 1
else
    redis.call('HMSET', KEYS[1], 'tokens', tokens, 'last', ARGV[3])
    return 0
end

Workers retry on 0 with a short asyncio.sleep.

Performance Tips

• Batch refill – instead of refilling on every take(), schedule a background task that updates all buckets every 100 ms. Reduces lock contention.
• Use asyncio.Queue for URL distribution – decouples URL production from consumption and naturally limits memory.
• Reuse aiohttp.ClientSession – connection pooling across proxies (via ProxyConnector) cuts TLS handshake overhead.
• Monitor bucket health – export tokens and refill_rate via Prometheus (gauge per proxy) to spot mis‑configured limits early.

Conclusion

A token‑bucket rate limiter gives you deterministic, per‑proxy throughput control without the guesswork of fixed sleeps. By coupling a lightweight TokenBucket class with a BucketManager keyed to proxy identifiers, you can drop the component into any async scraper, scale it across machines with Redis, and keep your IP reputation intact. The pattern works equally well for SEO SERP monitoring, price‑intelligence pipelines, or ad‑verification bots – wherever you need high‑volume, polite traffic from a rotating proxy fleet.

Give the code a spin, adjust capacity and refill_rate to match your provider’s limits, and watch the 429 errors disappear.