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.

The Bucket, Drawn

The whole algorithm is two numbers and a clock: how many tokens are in the bucket, and when it was last refilled. There is no timer thread and no scheduled task — the refill is computed lazily on each acquisition from the elapsed time, which is what makes the implementation both cheap and safe to share across coroutines.

Refill, capacity ceiling and consumption A bucket in the centre holds tokens. An inflow arrow on the left adds tokens at a fixed rate per second. A capacity line across the top shows that tokens above the ceiling are discarded rather than accumulated. An outflow arrow on the right removes one token per request, and a note explains that an empty bucket makes the caller wait for the next refill instant. refill_rate = max_rate / period tokens added lazily, on read capacity — overflow is discarded tokens available −1 per request geocode call proceeds immediately Empty bucket: the caller sleeps until the next token is due — computed, not polled — so an idle worker costs nothing.

The capacity line is the parameter people set carelessly. Capacity equal to the per-second rate gives you almost no burst tolerance and turns every scheduling hiccup into a stall; capacity set to a full minute’s worth lets an idle pipeline dump sixty seconds of traffic in one instant and trip the provider’s own limiter. Capacity in the range of two to five seconds of rate is the band that behaves well in practice.

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

Where the Lock Actually Matters

The threaded and async variants of this limiter look similar and guard against different hazards. In the threaded version, two OS threads can genuinely read self._tokens at the same instant, so the mutation must be inside a Lock. In the async version, no two coroutines run simultaneously — but any await inside the critical section is a suspension point where another coroutine will run, so the update must be free of awaits.

Two concurrency models, two different guards Two panels. The threaded panel shows two worker threads contending for a mutex around the refill-and-decrement step, with the note that true parallelism makes the lock mandatory. The async panel shows two coroutines interleaving on one thread, with the note that the refill-and-decrement step must contain no await, and that the sleep must happen after the token count has been updated. Threads — real parallelism worker thread A worker thread B with self._lock: refill from clock decrement one token Both threads can execute the read-modify-write at the same instant, so the mutex is not optional — without it the bucket over-issues under load. Asyncio — cooperative coroutine A coroutine B no await inside refill, decrement, compute the wait One thread, so no mutex is needed — but an await between reading and writing the count lets another coroutine observe stale state. Sleep afterwards.

The bug this prevents is subtle and only appears under concurrency: if the sleep happens before the decrement, every waiting coroutine wakes, sees the same replenished count, and all of them proceed. Decrement first, then sleep for the computed interval, and the bucket stays honest no matter how many callers are queued behind it.

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

Sizing the Limiter Against a Real Quota

Choosing max_rate is not the same as reading the provider’s published number. The published number is an upper bound enforced with their clock, their network jitter and their definition of a window; yours differs on all three counts. The safety margin below is what keeps a batch from tripping a limit it is nominally under.

From published quota to configured rate A four-step reduction. The published quota of 600 requests per minute is reduced by a five percent clock-skew margin, then by a ten percent retry headroom allowance, then divided across three concurrent jobs, arriving at a configured limiter rate of 170 requests per minute per job. published quota 600 / min their clock −5% clock skew 570 / min window edges disagree −10% retries 513 / min retries are billable calls ÷ 3 jobs 170 / min configured per job The final division is the step teams skip. Three jobs each configured at the full quota will collectively exceed it by three times, and the symptom — sporadic 429s under no obvious load — is genuinely hard to diagnose from one job's logs. If job count varies, keep the bucket in Redis and let every worker draw from one shared allowance instead.

That last line is the escape hatch when static division stops fitting: a shared counter in Redis, decremented with a small Lua script so the read and write are atomic, gives every worker a view of the same bucket. It costs a network round trip per acquisition, which is negligible next to a geocoding call, and it removes the coordination problem entirely.

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.