Implementing Token-Bucket Rate Limiting in Python

A token-bucket rate limiter holds a fixed number of tokens that refill at a steady rate; each geocoding request spends one token and blocks until a token is available, so your batch job sustains the provider’s average requests-per-second while still absorbing short bursts. This page is part of the Rate Limiting Strategies for Batch Processing guide.

The Algorithm

A token bucket is defined by four pieces of state and one rule. Tokens accumulate at rate per second up to a maximum of capacity; an acquire(n) call succeeds only when at least n tokens are present, subtracting them, otherwise the caller waits for the deficit to refill. The critical design choice is lazy refill against a monotonic clock: rather than a background thread topping the bucket up on a timer, you compute how many tokens should have arrived since the last acquire and add them on demand.

tokens_now = min(capacity, tokens_last + rate * (now - last_refill))
wait_for(n) = 0                       if tokens_now >= n
            = (n - tokens_now) / rate otherwise

Use time.monotonic() — never time.time() — for now. A wall-clock jump (NTP correction, daylight-saving change) would otherwise make elapsed negative or huge and either stall the bucket or release a flood of requests.

Parameter Breakdown

Parameter Meaning How to set it for a geocoding provider
rate Tokens added per second (sustained throughput) Provider’s sustained requests-per-second, minus ~5% safety margin
capacity Maximum tokens the bucket holds (burst allowance) The provider’s documented burst size, or 1 for strict smoothing
tokens Current token count (mutable state) Initialize to capacity so the first burst is allowed immediately
last_refill Monotonic timestamp of the previous refill Set to time.monotonic() at construction
n (per acquire) Tokens a single call costs 1 for one request; raise it for batch endpoints that count N addresses

Refill rate is the long-term ceiling; capacity only controls how bursty the short term is allowed to be. A provider that permits 10 requests/second with a 20-request burst maps to rate=9.5, capacity=20.

Thread-Safe Synchronous Limiter

The synchronous version guards its state with a threading.Lock so a pool of worker threads shares one quota. acquire blocks the calling thread until it can spend the tokens.

"""token_bucket.py — a thread-safe, monotonic-clock token bucket."""
from __future__ import annotations

import threading
import time


class TokenBucket:
    """Thread-safe token-bucket rate limiter.

    Args:
        rate: Sustained tokens granted per second.
        capacity: Maximum tokens the bucket can hold (burst allowance).

    Raises:
        ValueError: If rate or capacity is not positive.
    """

    def __init__(self, rate: float, capacity: float) -> None:
        if rate <= 0 or capacity <= 0:
            raise ValueError("rate and capacity must be positive")
        self._rate = float(rate)
        self._capacity = float(capacity)
        self._tokens = float(capacity)
        self._last = time.monotonic()
        self._lock = threading.Lock()

    def _refill_locked(self) -> None:
        """Add tokens accrued since the last refill. Caller holds the lock."""
        now = time.monotonic()
        elapsed = now - self._last
        if elapsed > 0:
            self._tokens = min(self._capacity, self._tokens + elapsed * self._rate)
            self._last = now

    def acquire(self, n: float = 1.0, timeout: float | None = None) -> bool:
        """Spend n tokens, blocking until they are available.

        Args:
            n: Tokens to consume (one per geocoding request by default).
            timeout: Max seconds to wait; None waits indefinitely.

        Returns:
            True if the tokens were acquired, False if timeout elapsed first.
        """
        if n > self._capacity:
            raise ValueError(f"cannot acquire {n} tokens; capacity is {self._capacity}")
        deadline = None if timeout is None else time.monotonic() + timeout
        while True:
            with self._lock:
                self._refill_locked()
                if self._tokens >= n:
                    self._tokens -= n
                    return True
                deficit = n - self._tokens
                wait = deficit / self._rate
            if deadline is not None:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    return False
                wait = min(wait, remaining)
            time.sleep(wait)

Sleeping outside the lock is deliberate: holding the lock during time.sleep would serialize all waiting threads and defeat the shared bucket.

Asyncio Limiter

For the event-loop dispatch used across async geocoding requests in Python, the same math runs inside an asyncio.Lock and yields with await asyncio.sleep so the loop stays free.

"""async_token_bucket.py — an asyncio-native token bucket."""
from __future__ import annotations

import asyncio
import time


class AsyncTokenBucket:
    """Asyncio token-bucket limiter usable as an async context manager.

    Args:
        rate: Sustained tokens granted per second.
        capacity: Maximum tokens the bucket can hold (burst allowance).
    """

    def __init__(self, rate: float, capacity: float) -> None:
        if rate <= 0 or capacity <= 0:
            raise ValueError("rate and capacity must be positive")
        self._rate = float(rate)
        self._capacity = float(capacity)
        self._tokens = float(capacity)
        self._last = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self, n: float = 1.0) -> None:
        """Await until n tokens are available, then spend them."""
        if n > self._capacity:
            raise ValueError(f"cannot acquire {n} tokens; capacity is {self._capacity}")
        while True:
            async with self._lock:
                now = time.monotonic()
                elapsed = now - self._last
                if elapsed > 0:
                    self._tokens = min(
                        self._capacity, self._tokens + elapsed * self._rate
                    )
                    self._last = now
                if self._tokens >= n:
                    self._tokens -= n
                    return
                wait = (n - self._tokens) / self._rate
            await asyncio.sleep(wait)

    async def __aenter__(self) -> "AsyncTokenBucket":
        await self.acquire()
        return self

    async def __aexit__(self, *exc: object) -> None:
        return None

Throttling an Async Batch

Wire the bucket in front of every outbound call. Sizing the bucket to the provider’s sustained requests-per-second is the whole job: set rate to the safe RPS and capacity to the burst you are allowed. Here a 10-RPS provider drains a batch smoothly.

import asyncio
import aiohttp

# 10 req/s sustained, allow a burst of 20 queued addresses.
BUCKET = AsyncTokenBucket(rate=9.5, capacity=20)


async def geocode_one(session: aiohttp.ClientSession, address: str) -> dict:
    """Geocode a single address, gated by the shared token bucket."""
    await BUCKET.acquire()
    async with session.get(
        "https://api.provider.example/v1/geocode",
        params={"q": address},
        timeout=aiohttp.ClientTimeout(total=10),
    ) as resp:
        resp.raise_for_status()
        return await resp.json()


async def geocode_batch(addresses: list[str]) -> list[dict]:
    """Geocode many addresses concurrently while the bucket enforces the RPS cap."""
    async with aiohttp.ClientSession() as session:
        tasks = [geocode_one(session, a) for a in addresses]
        return await asyncio.gather(*tasks, return_exceptions=True)

Vectorized pandas variant

When the input is a DataFrame, collect the address column and run the throttled batch once per frame — never call asyncio.run per row, which destroys the event loop (and the bucket’s timing) each time.

import asyncio
import pandas as pd


def geocode_dataframe(df: pd.DataFrame, address_col: str = "address") -> pd.DataFrame:
    """Geocode df[address_col] through the token bucket and attach results.

    Args:
        df: Input frame of raw addresses.
        address_col: Column holding address strings.

    Returns:
        A copy of df with a 'geocode_result' column in input order.
    """
    addresses: list[str] = df[address_col].astype(str).tolist()
    results = asyncio.run(geocode_batch(addresses))
    out = df.copy()
    out["geocode_result"] = results
    return out

Edge Cases

Requesting more tokens than capacity

An acquire(n) where n > capacity can never be satisfied — the bucket physically cannot hold that many tokens, so the caller would wait forever. Both classes raise ValueError up front. If a batch endpoint counts, say, 100 addresses per call, capacity must be at least 100.

Clock choice under NTP correction

time.time() can step backward when the system clock is corrected, producing a negative elapsed that silently stalls the bucket, or a large forward jump that dumps a burst of tokens and triggers 429s. Using time.monotonic() (as both classes do) makes elapsed strictly non-negative and immune to wall-clock adjustments.

Fractional refill starvation

With a very low rate (for example a 60 RPM provider, rate=1.0), a caller requesting n=1 waits up to a full second per token. That is correct, but if many threads contend, verify the sleep happens outside the lock (it does above) so waiters wake in roughly arrival order rather than all re-contending on the held lock.

Integration Note

A token bucket is the pre-emptive half of throttling; pair it with the reactive backoff and circuit-breaker patterns in the parent rate limiting strategies for batch processing guide so that a bucket sized slightly too high still degrades gracefully on a 429. When you scale beyond one provider or one machine, the bucket becomes one lever in optimizing batch geocoding throughput, where connection pooling and chunking determine whether you actually reach the RPS the bucket permits.