Handling API Rate Limits with Token-Bucket Throttling #

A batch deposit run that fans out fifty concurrent workers against a repository API is, from the server’s point of view, a small denial-of-service event. Zenodo, DataCite, and figshare all publish per-token request ceilings, and the moment your workers cross one the server stops returning 201 Created and starts returning 429 Too Many Requests — often with a Retry-After header that your code must obey or be throttled harder. This page is the concrete limiter that sits underneath the concurrency model in the Async Batch Processing overview: where that guide bounds how many coroutines run at once, a token bucket bounds how fast they are allowed to call out, independent of how many of them there are. It assumes you already run an asyncio deposit loop and are comfortable with Python 3.10+; the goal is to replace per-worker sleeps and naive retry loops with a single, shared, rate-accurate gate.

The distinction matters because a semaphore and a rate limiter solve different problems. A semaphore of fifty permits still lets fifty requests leave in the same millisecond, then fifty more the instant those return — the instantaneous rate is unbounded even though concurrency is capped. Servers rate-limit on requests-per-second across your whole token, not on your concurrency, so the correct control is a token bucket shared by every worker: a bucket refills at a fixed global rate, each request must remove a token before it goes out, and a worker with no token available waits. When the server does push back with a 429, the same bucket is the natural place to absorb the Retry-After pause so that all workers back off together rather than each discovering the limit independently.

The Parameter Reference Table #

A token bucket has exactly three tuning knobs plus one server-supplied override, and every one of them has a concrete failure mode when set wrong. The rate is the sustained throughput the server tolerates; capacity is how many tokens the bucket can hold, which is the same thing as the maximum instantaneous burst; and capacity divided by rate is the longest quiet period after which you may fire a full burst. The Retry-After value is not a knob you choose — it is the server telling you exactly how long to stop — and the single most common production incident is ignoring it. The table below is the authoritative reference for sizing a limiter against a real repository quota.

The rule of thumb for a shared limiter is to set rate to roughly 80% of the documented quota (leaving headroom for clock skew and other clients on the same token), and capacity to somewhere between one and three seconds’ worth of tokens — enough to absorb a normal retry spike, small enough that a long idle period cannot accumulate a burst large enough to breach a per-second sub-limit.

Production Implementation #

The limiter below is a single object shared by every worker. Refill is computed lazily from a monotonic clock on each acquire rather than driven by a background task, so there is no timer coroutine to leak and the bucket is correct even if the event loop was starved. The pause method implements Retry-After by draining the bucket and pushing its refill clock into the future, which forces every waiting worker to block until the server’s cooldown elapses — the coordinated backoff a per-worker sleep cannot give you. This is the throttling layer beneath the batch deposit automation described in the Repository Deposit Automation guide, and it applies unchanged to the bulk metadata operations in batch updating DataCite metadata for existing DOIs.

from __future__ import annotations

import asyncio
import logging
import random
import time
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone

import httpx

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("fair.throttle")


class TokenBucket:
    """A rate limiter shared across coroutines. Refill is lazy and clock-based."""

    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            # tokens added per second
        self._capacity: float = capacity    # maximum tokens held == burst size
        self._tokens: float = capacity      # start full so a warm run bursts once
        self._updated: float = time.monotonic()
        self._lock: asyncio.Lock = asyncio.Lock()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._updated
        # elapsed can be negative while a Retry-After pause is in effect.
        self._tokens = max(0.0, min(self._capacity, self._tokens + elapsed * self._rate))
        self._updated = now

    async def acquire(self, tokens: float = 1.0) -> None:
        """Block until `tokens` are available, then consume them."""
        if tokens > self._capacity:
            raise ValueError(f"cannot acquire {tokens}: exceeds capacity {self._capacity}")
        while True:
            async with self._lock:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return
                deficit = tokens - self._tokens
                wait = deficit / self._rate
            await asyncio.sleep(wait)  # released the lock before sleeping

    async def pause(self, seconds: float) -> None:
        """Honor a server Retry-After: drain the bucket and hold all workers."""
        async with self._lock:
            self._tokens = 0.0
            # Push the refill clock forward so no tokens accrue until the cooldown ends.
            self._updated = time.monotonic() + seconds
        logger.warning("bucket paused for %.2fs by server backpressure", seconds)


def parse_retry_after(value: str | None, *, default: float = 5.0) -> float:
    """Parse a Retry-After header: integer seconds OR an HTTP-date."""
    if not value:
        return default
    value = value.strip()
    if value.isdigit():
        return float(value)
    try:  # RFC 7231 HTTP-date form
        when = parsedate_to_datetime(value)
        if when.tzinfo is None:
            when = when.replace(tzinfo=timezone.utc)
        delta = (when - datetime.now(timezone.utc)).total_seconds()
        return max(0.0, delta)
    except (TypeError, ValueError):
        return default


async def deposit_one(
    client: httpx.AsyncClient,
    url: str,
    payload: dict[str, object],
    bucket: TokenBucket,
    *,
    max_retries: int = 5,
    base_delay: float = 1.0,
) -> dict[str, object]:
    """Deposit one record through the shared limiter, honoring Retry-After."""
    for attempt in range(max_retries + 1):
        await bucket.acquire()
        try:
            resp = await client.post(url, json=payload, timeout=30.0)
        except (httpx.ConnectError, httpx.ReadTimeout) as exc:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            logger.warning("transport error %s; retry %d in %.2fs", exc, attempt + 1, delay)
            await asyncio.sleep(delay)
            continue

        if resp.status_code == 429 or resp.status_code == 503:
            cooldown = parse_retry_after(resp.headers.get("Retry-After"))
            await bucket.pause(cooldown)
            # Full jitter on top of the server value so workers do not re-enter in lockstep.
            await asyncio.sleep(cooldown + random.uniform(0, base_delay))
            continue
        if resp.status_code >= 500:
            if attempt == max_retries:
                resp.raise_for_status()
            await asyncio.sleep(base_delay * (2 ** attempt) + random.uniform(0, 0.5))
            continue

        resp.raise_for_status()  # 4xx other than 429 is terminal
        return resp.json()

    raise RuntimeError(f"deposit exhausted {max_retries} retries for {url}")


async def run_batch(
    records: list[dict[str, object]], url: str, *, rate: float = 8.0, capacity: float = 16.0
) -> list[dict[str, object]]:
    """Fan out concurrent deposits sharing ONE bucket, so the global rate holds."""
    bucket = TokenBucket(rate=rate, capacity=capacity)
    async with httpx.AsyncClient() as client:
        tasks = [deposit_one(client, url, rec, bucket) for rec in records]
        return await asyncio.gather(*tasks)

Verification #

You do not need a live repository to prove the limiter is correct: assert that the bucket enforces its rate, and that a 429 with a Retry-After header stalls every worker for the mandated cooldown. The httpx.MockTransport below returns one 429 then a 201, and the test asserts the observed gap is at least the cooldown the server demanded.

import asyncio
import time

import httpx
import pytest


@pytest.mark.asyncio
async def test_bucket_enforces_rate() -> None:
    """A bucket of capacity 2 at 2 tokens/s must space 4 acquires by >= ~1s total."""
    bucket = TokenBucket(rate=2.0, capacity=2.0)
    start = time.monotonic()
    for _ in range(4):
        await bucket.acquire()
    elapsed = time.monotonic() - start
    assert elapsed >= 0.9  # 2 free from the full bucket, 2 more at 2/s


@pytest.mark.asyncio
async def test_retry_after_is_honored() -> None:
    calls: list[float] = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(time.monotonic())
        if len(calls) == 1:
            return httpx.Response(429, headers={"Retry-After": "1"})
        return httpx.Response(201, json={"id": "10.5072/abc"})

    bucket = TokenBucket(rate=100.0, capacity=100.0)
    transport = httpx.MockTransport(handler)
    async with httpx.AsyncClient(transport=transport, base_url="https://api.test") as client:
        result = await deposit_one(client, "/deposit", {"title": "x"}, bucket)

    assert result["id"] == "10.5072/abc"
    assert calls[1] - calls[0] >= 1.0  # the server-mandated cooldown was observed

Run it with pytest -q test_throttle.py. A passing run prints 2 passed; the test_retry_after_is_honored case fails loudly if you ever regress to ignoring the header, because the second call arrives too soon.

Gotchas #

• One bucket per worker is not a shared limiter. Constructing a TokenBucket inside each coroutine gives every worker its own private rate, so N workers deposit at N times the configured rate and breach the quota instantly. Fix: build a single bucket in the runner and pass the same instance to every worker, as run_batch does.
• Sleeping while holding the lock stalls every worker. If acquire awaits asyncio.sleep inside the async with self._lock block, one waiting worker blocks all the others from even checking the bucket, collapsing throughput to serial. Fix: compute the wait time under the lock, release it, and only then sleep — the ordering the implementation uses.
• Subtracting jitter from Retry-After re-trips the limit. Treating the server cooldown as a target to jitter around means some workers re-enter early and immediately earn another 429, which servers often escalate into a longer ban. Fix: pause the bucket for the full server value and add jitter only as extra delay on top.

Frequently Asked Questions #

Why use a token bucket instead of an asyncio.Semaphore? #

They control different axes. A semaphore caps concurrency — the number of requests in flight simultaneously — but places no limit on how quickly permits are released and re-acquired, so fifty workers can still exceed a requests-per-second quota in bursts. A token bucket caps the rate, which is what repository APIs actually meter. In practice you use both: a semaphore to bound open sockets and memory, and one shared token bucket to keep the aggregate call rate under the server’s ceiling.

How should jitter interact with a server Retry-After value? #

Treat Retry-After as a hard floor and add jitter on top, never inside it. Every worker must wait at least the full cooldown the server demanded, so the bucket is paused for exactly that long; the extra random.uniform(0, base_delay) you sleep afterwards only staggers re-entry so the workers do not all fire the instant the cooldown expires. Subtracting jitter from the cooldown would re-trip the limit and can escalate a soft 429 into a longer ban.

Related Guides #

• Async Batch Processing — the parent overview whose bounded-concurrency worker pool this limiter throttles by rate.
• Repository Deposit Automation — the deposit workflows that call repository APIs under exactly these quotas.
• Batch Updating DataCite Metadata for Existing DOIs — a bulk-update job where a shared bucket keeps thousands of PUTs under the DataCite rate ceiling.