Rate-Limiting Concurrent Requests in Python CLIs

You parallelised a command with sixteen workers and it went from four minutes to fifteen seconds — for the first run. Then the API started answering 429 Too Many Requests, your retries kicked in, the retries got 429s too, and the command ended up slower than the sequential version, with a warning email from the platform team about your token. Concurrency multiplies your request rate, and most APIs enforce a quota: so many requests per second, per minute or per hour. The fix is not fewer workers — it is pacing. This guide explains the difference between limiting concurrency and limiting rate, implements a token bucket for both thread pools and asyncio, and shows how to derive sensible settings and expose them to users. It is part of the concurrency and async topic.

Prerequisites

• Python 3.11+ and httpx.
• A concurrent command built on a thread pool or asyncio.
• The API's documented limits. If none are documented, the response headers usually reveal them.

Two different limits

A pool size or a semaphore limits how many requests are in flight at once. A rate limiter limits how many requests start per unit of time. They are not interchangeable, and APIs frequently enforce both.

Consider eight workers against an API allowing 10 requests per second. If each request takes 400 ms, eight in flight produce about 20 requests per second — double the quota — even though concurrency is modest. If the server speeds up to 100 ms per request, the same eight workers produce 80 per second. A semaphore alone cannot protect a quota, because your rate depends on the server's latency. Conversely, a rate limit alone can let slow requests pile up into hundreds of open connections. Use a semaphore (or pool size) for the concurrency bound and a rate limiter for the quota.

The token bucket

The token bucket is the standard pacing algorithm because it allows short bursts while enforcing an average. A bucket holds up to capacity tokens and refills at rate tokens per second; each request takes a token, waiting when the bucket is empty.

With rate=10 and capacity=5, a command can fire five requests instantly at start-up, then settles to one every 100 ms. Keeping capacity small avoids a large burst that some APIs count against a per-second window.

The recipe

One implementation serves threads; a thin async variant serves asyncio. Clock and sleep are injected so the behaviour can be tested exactly.

# src/mytool/ratelimit.py
from __future__ import annotations

import asyncio
import threading
import time
from collections.abc import Callable


class TokenBucket:
    """Thread-safe token bucket: `rate` tokens/second, up to `capacity`."""

    def __init__(self, rate: float, capacity: float | None = None, *,
                 clock: Callable[[], float] = time.monotonic,
                 sleep: Callable[[float], None] = time.sleep) -> None:
        if rate <= 0:
            raise ValueError("rate must be positive")
        self.rate = rate
        self.capacity = capacity if capacity is not None else max(1.0, rate)
        self._tokens = self.capacity
        self._clock = clock
        self._sleep = sleep
        self._last = clock()
        self._lock = threading.Lock()

    def _refill(self) -> None:
        now = self._clock()
        self._tokens = min(self.capacity, self._tokens + (now - self._last) * self.rate)
        self._last = now

    def reserve(self) -> float:
        """Take a token; return how long the caller must wait before using it."""
        with self._lock:
            self._refill()
            self._tokens -= 1
            return 0.0 if self._tokens >= 0 else -self._tokens / self.rate

    def acquire(self) -> None:
        wait = self.reserve()
        if wait > 0:
            self._sleep(wait)

    def slow_down(self, factor: float = 0.5) -> None:
        """Reduce the rate after a 429 (never below 0.1/s)."""
        with self._lock:
            self.rate = max(0.1, self.rate * factor)


class AsyncTokenBucket(TokenBucket):
    async def acquire_async(self) -> None:
        wait = self.reserve()
        if wait > 0:
            await asyncio.sleep(wait)

reserve() is the heart of it. Instead of looping "check, sleep, check again", it takes the token immediately — letting the balance go negative — and tells the caller exactly how long to wait. Each caller thereby queues behind the ones before it, in order, with one lock acquisition and no busy-waiting. The lock is held only for arithmetic, never while sleeping.

Wiring it into a thread-pool command takes one line in the worker, alongside the pool size that bounds concurrency:

# src/mytool/cli.py
from concurrent.futures import ThreadPoolExecutor, as_completed

import httpx
import typer

from mytool.ratelimit import TokenBucket

app = typer.Typer()


@app.callback()
def main() -> None:
    """Issue tracker tools."""


@app.command()
def export(
    ids: list[int],
    jobs: int = typer.Option(8, "--jobs", "-j", min=1, max=32),
    rate: float = typer.Option(10.0, "--rate", min=0.1, help="Max requests per second."),
) -> None:
    """Fetch each issue by ID, staying under the API's rate limit."""
    bucket = TokenBucket(rate=rate, capacity=min(jobs, rate))
    with httpx.Client(base_url="https://tracker.example.com/api",
                      limits=httpx.Limits(max_connections=jobs)) as client:

        def fetch(issue_id: int) -> tuple[int, int]:
            bucket.acquire()
            r = client.get(f"/issues/{issue_id}")
            if r.status_code == 429:
                bucket.slow_down()
            return issue_id, r.status_code

        with ThreadPoolExecutor(max_workers=jobs) as pool:
            futures = [pool.submit(fetch, i) for i in ids]
            results = sorted(f.result() for f in as_completed(futures))
    limited = sum(1 for _, code in results if code == 429)
    typer.echo(f"fetched {len(results)} issues, {limited} rate-limited", err=True)


if __name__ == "__main__":
    app()

In asyncio, the same bucket works with await bucket.acquire_async() inside the semaphore-guarded task. Because reserve() holds a threading.Lock only briefly and never awaits while holding it, it is safe to call from coroutines on one event loop.

Choosing the numbers

Start from the documented limit and stay a little under it — 80–90% leaves room for other clients sharing the same token and for clock differences between you and the server. Then the maths is simple: the minimum time for N requests is N / rate, and concurrency beyond rate × typical latency adds nothing but idle connections.

Many APIs publish the live budget in headers — X-RateLimit-Limit, X-RateLimit-Remaining and X-RateLimit-Reset, or the standardised RateLimit-Policy and RateLimit headers. A well-behaved client reads them: if Remaining is low and Reset is far off, slow the bucket down; if a 429 arrives with Retry-After, sleep for it, as shown in retries and backoff for CLI HTTP calls. The slow_down() method above is the simplest adaptive step: halve the rate on each 429.

UX considerations

• Expose --rate alongside --jobs. Users with a higher quota, or a shared token that others are using too, need to tune it. Document the default and the API limit it came from.
• Explain slow progress. If a large export is pacing at 5 requests per second, show an ETA and say why: "rate-limited to 5 req/s (API quota); ~4 minutes remaining". Users accept a slow tool they understand.
• Report 429s in the summary. A count of rate-limited responses tells users when the default rate is too aggressive for their account.
• Keep one limiter per quota. If several commands in one process hit the same API, share one bucket (for example on the Click context) rather than giving each command its own, or they will jointly exceed the quota.
• Remember other processes. A token bucket paces one process. Two terminals running the same command double the rate; for heavy tools, file locking to allow only one export at a time can be the simplest guard.

Testing the behaviour

With an injected clock, you can assert the exact wait each caller receives — no sleeping, no flaky timing:

# tests/test_ratelimit.py
import asyncio

import pytest

from mytool.ratelimit import AsyncTokenBucket, TokenBucket


class FakeTime:
    def __init__(self) -> None:
        self.now = 0.0
        self.slept: list[float] = []

    def clock(self) -> float:
        return self.now

    def sleep(self, s: float) -> None:
        self.slept.append(round(s, 6))
        self.now += s


def test_burst_then_steady_rate():
    t = FakeTime()
    bucket = TokenBucket(rate=10, capacity=3, clock=t.clock, sleep=t.sleep)
    waits = [round(bucket.reserve(), 6) for _ in range(6)]
    assert waits == [0.0, 0.0, 0.0, 0.1, 0.2, 0.3]


def test_refills_over_time():
    t = FakeTime()
    bucket = TokenBucket(rate=2, capacity=2, clock=t.clock, sleep=t.sleep)
    bucket.acquire(); bucket.acquire()
    t.now += 1.0                    # one second later: two tokens back
    assert bucket.reserve() == 0.0
    assert bucket.reserve() == 0.0


def test_average_rate_is_enforced():
    t = FakeTime()
    bucket = TokenBucket(rate=5, capacity=1, clock=t.clock, sleep=t.sleep)
    for _ in range(51):
        bucket.acquire()
    assert t.now == pytest.approx(10.0)   # 50 intervals of 0.2 s


def test_slow_down_halves_rate():
    bucket = TokenBucket(rate=8)
    bucket.slow_down()
    assert bucket.rate == 4


def test_async_variant_paces():
    bucket = AsyncTokenBucket(rate=100, capacity=1)

    async def run() -> float:
        loop = asyncio.get_running_loop()
        start = loop.time()
        for _ in range(11):
            await bucket.acquire_async()
        return loop.time() - start

    assert asyncio.run(run()) >= 0.09

The first test captures the token bucket's defining behaviour — a burst up to capacity, then evenly spaced waits — as a precise contract. The async test uses a real clock with a high rate so it completes in about a tenth of a second.

Conclusion

Concurrency decides how many requests are in flight; a rate limiter decides how many start each second. For API-backed CLIs you usually need both: a pool or semaphore sized with --jobs, and a token bucket set just under the documented quota with --rate. Implement the bucket with a reservation so callers queue fairly without busy-waiting, react to 429s by slowing down, and test it with a fake clock. Your command then runs as fast as the API allows — and no faster.

Frequently asked questions

Related

• Up: Concurrency and async in Python CLIs
• Parallelising CLI work with thread pools
• Running async code in Typer and Click
• Retries and backoff for CLI HTTP calls
• Paginating API results in a CLI