Implementing Token-Bucket Throttling for Grants.gov APIs

Async token-bucket throttle for Grants.gov APIs in Python: monotonic-clock refill, httpx wrapper, Retry-After and X-RateLimit feedback, 2 CFR 200.302 audit logging.

This guide is part of the API Polling & Rate Limiting section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: how do you pace a fleet of concurrent Grants.gov API calls so you stay under the published limit, absorb the server’s own throttling signals, and leave a replayable record of every pacing decision?

The naive answer — asyncio.sleep(0.5) between requests, or a bare semaphore of size ten — throttles the wrong thing. A fixed sleep wastes headroom when the bucket is full and still overruns when a burst of workers wakes at once; a semaphore caps concurrency but not rate, so ten fast handlers still hammer the endpoint. A token bucket separates the two: it allows short bursts up to a capacity while enforcing a long-run average, and it gives you one place to fold the server’s Retry-After and X-RateLimit-* headers back into your own pacing.

When to Use This Approach

Reach for a token-bucket throttle when all three conditions hold:

  • You issue many concurrent requests against one shared limit. Grants.gov meters by API key, not by connection, so every coroutine in your event loop draws from the same budget. A limiter that lives beside the HTTP client — not inside each task — is the only way to keep the aggregate rate honest.
  • The calls feed a regulated pipeline. Every external retrieval that lands a grant record into your system is an event you must be able to reconstruct. 2 CFR §200.302 requires financial-management records to identify the source of every transaction, so each throttling decision — dispatched, delayed, or deferred by the server — is logged with the token cost and the wait it incurred.
  • The server talks back. Grants.gov returns 429 Too Many Requests with a Retry-After, and healthy responses carry X-RateLimit-Remaining and X-RateLimit-Reset. A limiter that ignores those headers will keep marching into the wall; this one treats them as the authoritative signal and clamps the local bucket to match.

Retry orchestration, backoff budgets, and fallback routing after a hard failure are explicitly out of scope here — the bucket only decides when a request may leave, not what to do when one fails. That recovery logic lives in Pipeline Fallback & Retry Logic. Deterministic result paging belongs to Paginating Grants.gov Search Results Deterministically, and the broader taxonomy of portal limit behaviors belongs to Handling Rate Limits in Grant Portal APIs.

Feedback-controlled token bucket pacing Grants.gov API calls A refill clock drips tokens into a token bucket at a fixed refill_rate up to its capacity. Each acquire() call tests whether at least one token is available. When tokens are present the request drains one and flows through an httpx AsyncClient to the Grants.gov API; the response headers are parsed and a dashed feedback arrow adjusts the bucket rate from Retry-After and X-RateLimit-Reset. When the bucket is empty the call branches to an await-refill node that sleeps for the token deficit divided by the refill rate and then wakes back into the bucket. Feedback-controlled token bucket Tokens refill at a fixed rate; each request drains one; server headers throttle the rate back down. refill_rate acquire() yes empty → await Retry-After / X-RateLimit-Reset → adjust rate wake on refill Refill clock time.monotonic() Token bucket capacity = N tokens burst headroom tokens ≥ 1? httpx.AsyncClient drain 1 · GET Grants.gov API Search2 endpoint Parse headers 429 · X-RateLimit-* Await refill sleep(deficit / rate)

Step-by-Step Implementation

The reference implementation targets Python 3.10+ and httpx for its native async client. Pin the dependencies so the pacing behavior is reproducible across CI and production:

bash
pip install "httpx==0.27.0"
# stdlib only otherwise: asyncio, time, logging, dataclasses

One contract to internalize before writing any code: the bucket measures time with time.monotonic(), never time.time(). Wall-clock time can jump backward on an NTP correction or forward across a DST change, which would either freeze the bucket or grant a windfall of phantom tokens. The monotonic clock only ever moves forward, so the refill math is stable regardless of what the system clock does.

Step 1: Model the token bucket with a monotonic-clock refill

Model the bucket as a small value object holding capacity, refill_rate (tokens per second), the current token count, and the monotonic timestamp of the last refill. Tokens are never added on a timer; instead they are computed lazily — every time you touch the bucket, you fold in elapsed * refill_rate and clamp at capacity. This “refill on read” trick means no background task and no drift.

python
import logging
import time
from dataclasses import dataclass, field

LOGGER = logging.getLogger("grants_gov.throttle")


class RateLimitError(RuntimeError):
    """Raised when a request cannot be satisfied within the limiter's bounds."""


@dataclass
class TokenBucket:
    """A monotonic-clock token bucket. Refills lazily on every read."""

    capacity: float
    refill_rate: float  # tokens per second
    _tokens: float = field(init=False)
    _updated: float = field(init=False)

    def __post_init__(self) -> None:
        if self.capacity <= 0 or self.refill_rate <= 0:
            raise ValueError("capacity and refill_rate must both be positive")
        self._tokens = self.capacity
        self._updated = time.monotonic()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._updated
        if elapsed <= 0:  # monotonic never goes backward, but guard anyway
            return
        self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_rate)
        self._updated = now

    @property
    def available(self) -> float:
        self._refill()
        return self._tokens

capacity sets the maximum burst — how many requests may leave back-to-back after an idle period — while refill_rate sets the sustained ceiling. For a documented limit of, say, two requests per second with a small allowance for bursts, TokenBucket(capacity=5, refill_rate=2.0) lets five queued calls fire immediately, then settles to two per second.

Step 2: Implement async acquire() that awaits capacity

acquire() is the gate every request passes through. Under an asyncio.Lock, it refills, and if a token is available it debits and returns immediately. If not, it computes the exact deficit, sleeps for deficit / refill_rate — the precise time until enough tokens accrue — and loops. Holding the lock across the sleep serializes waiters into FIFO order, which keeps dispatch deterministic instead of a thundering herd.

python
import asyncio
from typing import Optional


class AsyncTokenBucket(TokenBucket):
    """Async-safe token bucket that awaits capacity rather than rejecting."""

    def __post_init__(self) -> None:
        super().__post_init__()
        self._lock = asyncio.Lock()

    async def acquire(self, tokens: float = 1.0, timeout: Optional[float] = None) -> float:
        """Block until `tokens` are available; return seconds waited."""
        if tokens > self.capacity:
            raise RateLimitError(
                f"request of {tokens} tokens exceeds bucket capacity {self.capacity}"
            )
        deadline = None if timeout is None else time.monotonic() + timeout
        waited = 0.0
        async with self._lock:
            while True:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return waited
                deficit = tokens - self._tokens
                delay = deficit / self.refill_rate
                if deadline is not None and time.monotonic() + delay > deadline:
                    raise RateLimitError(
                        f"acquire timed out: need {deficit:.2f} tokens, "
                        f"{delay:.2f}s exceeds remaining budget"
                    )
                LOGGER.debug("bucket empty; sleeping %.3fs for %.2f tokens", delay, deficit)
                await asyncio.sleep(delay)
                waited += delay

The returned waited value is not cosmetic: it is the throttling cost of this specific call, and Step 5 logs it so an auditor can see exactly how long the limiter held each request. An oversized request — more tokens than the bucket can ever hold — fails fast with a RateLimitError rather than deadlocking forever.

Step 3: Wrap httpx.AsyncClient with the limiter

Bind one bucket to one httpx.AsyncClient inside a thin wrapper. Every outbound call first awaits a token, then dispatches. Because the bucket is shared by reference, a hundred coroutines calling client.get(...) concurrently all draw from the same budget — the aggregate rate stays under the ceiling no matter how many tasks you fan out.

python
import httpx


class ThrottledGrantsClient:
    """httpx.AsyncClient wrapper that paces every request through a token bucket."""

    def __init__(self, base_url: str, bucket: AsyncTokenBucket, timeout: float = 30.0) -> None:
        self._bucket = bucket
        self._client = httpx.AsyncClient(base_url=base_url, timeout=timeout)

    async def get(self, url: str, **kwargs: object) -> httpx.Response:
        waited = await self._bucket.acquire(1.0)
        LOGGER.info(
            "dispatch GET %s | waited=%.3fs | tokens_left=%.2f",
            url, waited, self._bucket.available,
        )
        response = await self._client.get(url, **kwargs)
        self._apply_server_feedback(response)
        return response

    async def __aenter__(self) -> "ThrottledGrantsClient":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self._client.aclose()

Exposing the wrapper as an async context manager guarantees the underlying connection pool is closed on exit; a leaked AsyncClient is a common source of “Unclosed client” warnings and socket exhaustion in long-running poll loops.

Step 4: Respect server rate-limit headers as feedback

Your bucket is a model of the server’s limit, and models drift. The server is the source of truth, so fold its headers back in after every response. A 429 with a Retry-After drains the bucket and pushes the next refill out by that many seconds; a healthy response with X-RateLimit-Remaining and X-RateLimit-Reset clamps the local token count to whatever the server says is actually left.

python
from email.utils import parsedate_to_datetime


class AdaptiveTokenBucket(AsyncTokenBucket):
    """Token bucket that can be corrected by authoritative server signals."""

    def penalize(self, retry_after: float) -> None:
        """Honor a 429 Retry-After: empty the bucket and defer the next refill."""
        self._refill()
        self._tokens = 0.0
        self._updated = time.monotonic() + max(retry_after, 0.0)

    def reconcile(self, server_remaining: float, reset_epoch: float) -> None:
        """Clamp local tokens to the server's advertised remaining budget."""
        self._refill()
        self._tokens = min(self._tokens, max(server_remaining, 0.0))
        window = reset_epoch - time.time()
        if window > 0 and server_remaining > 0:
            observed_rate = server_remaining / window
            self.refill_rate = min(self.refill_rate, observed_rate)


def _parse_retry_after(raw: Optional[str]) -> Optional[float]:
    """Retry-After is either delta-seconds or an HTTP-date; return seconds."""
    if raw is None:
        return None
    try:
        return float(raw)  # delta-seconds form
    except ValueError:
        try:
            when = parsedate_to_datetime(raw)
            return max((when.timestamp() - time.time()), 0.0)
        except (TypeError, ValueError) as exc:
            LOGGER.warning("unparseable Retry-After %r: %s", raw, exc)
            return None

Wire these onto the client so _apply_server_feedback routes each response to the right correction:

python
    def _apply_server_feedback(self, response: httpx.Response) -> None:
        headers = response.headers
        if response.status_code == 429:
            retry_after = _parse_retry_after(headers.get("Retry-After"))
            if retry_after is not None:
                self._bucket.penalize(retry_after)
                LOGGER.warning("429 from Grants.gov | honoring Retry-After=%.1fs", retry_after)
            return
        remaining = headers.get("X-RateLimit-Remaining")
        reset = headers.get("X-RateLimit-Reset")
        if remaining is not None and reset is not None:
            try:
                self._bucket.reconcile(float(remaining), float(reset))
            except ValueError as exc:
                LOGGER.warning("malformed X-RateLimit headers: %s", exc)

Note that reconcile only ever lowers refill_rate — it takes the min of the configured rate and the observed one. The server can tell you to slow down; it can never trick your client into speeding past its own configured ceiling.

Step 5: Log throttling decisions for the audit trail

Every pacing decision is an event that touched an external federal system, so it belongs in the audit trail that 2 CFR §200.302 expects for records that identify the source of each transaction. Configure the module logger once, at process start, to emit structured lines to a write-once sink, and let the INFO dispatch line plus the WARNING penalty line carry the full story: what was requested, how long it waited, and whether the server pushed back.

python
def configure_throttle_logging(stream=None) -> None:
    """Attach a structured handler to the throttle logger exactly once."""
    if LOGGER.handlers:
        return
    handler = logging.StreamHandler(stream)
    handler.setFormatter(
        logging.Formatter("%(asctime)s | %(levelname)s | grants_gov.throttle | %(message)s")
    )
    LOGGER.addHandler(handler)
    LOGGER.setLevel(logging.INFO)

Ship those lines to an append-only tier. A dispatched call logs at INFO with its wait and remaining tokens; a server-driven slowdown logs at WARNING with the honored delay. Together they let a reviewer reconstruct the exact pacing of any polling window — the proof that your automation stayed within the funder’s published limits.

Verification

Confirm the limiter behaves deterministically with four checks:

  1. The sustained rate is capped. Drain the burst capacity, then fire more calls than the refill can supply and assert the total elapsed time is at least extra_requests / refill_rate. This proves acquire() actually blocks rather than passing through.
  2. Bursts are allowed up to capacity. From a full bucket, assert the first capacity acquisitions return a waited of 0.0 — the headroom is spent before any sleep occurs.
  3. A 429 defers the next dispatch. Call penalize(2.0) and assert available stays at 0.0 until roughly two seconds of monotonic time have passed, confirming the deferred _updated timestamp works.
  4. Server headers only slow you down. Call reconcile with a low server_remaining and a near reset window, and assert refill_rate never rose above its configured value and _tokens was clamped down.

A compliant run emits one INFO audit line per dispatch and one WARNING per honored Retry-After; nothing is silently dropped.

python
import asyncio

async def _smoke() -> None:
    bucket = AdaptiveTokenBucket(capacity=3, refill_rate=2.0)
    first_three = [await bucket.acquire(1.0) for _ in range(3)]
    assert all(w == 0.0 for w in first_three)          # burst is free
    slowed = await bucket.acquire(1.0)
    assert slowed >= 0.4                                # 4th call waits ~1/rate
    bucket.penalize(2.0)
    assert bucket.available == 0.0                      # 429 empties the bucket

asyncio.run(_smoke())

Common Errors & Fixes

Error Cause Fix
Aggregate rate exceeds the limit under load A separate bucket per task, so each has its own budget Construct one AdaptiveTokenBucket and share it by reference across every ThrottledGrantsClient call.
Bucket freezes or grants phantom tokens after a clock change Refill math built on time.time() wall-clock Use time.monotonic() for all bucket timestamps; reserve time.time() only for comparing against epoch reset headers.
acquire() never returns Requested more tokens than capacity, or refill_rate is zero Validate positivity in __post_init__ and raise RateLimitError when tokens > capacity instead of looping forever.
Repeated 429s despite local pacing Retry-After and X-RateLimit-* headers ignored Route every response through _apply_server_feedback; penalize on 429, reconcile on the remaining/reset pair.
RuntimeError: Event loop is closed / unclosed client warnings httpx.AsyncClient never closed Use the wrapper as an async context manager (async with ThrottledGrantsClient(...)) so aclose() always runs.