Code Snippet — August 4, 2026 ### Python: Token Bucket Rate Limiter with

🔧 Code Snippet — August 4, 2026

Python: Token Bucket Rate Limiter with Burst, Async & Metrics

rate limiting is too dumb. This gives you burst capacity (consume fast, refill steady), per-key limits (different quotas per user/API key), both sync & async , wait mode, exception mode, and live metrics for monitoring. Zero deps — pure Python 3.8+.

import time, threading
from dataclasses import dataclass, field

class TokenBucketRateLimiter:
    def __init__(self, rate, capacity=None, per_key=False):
        self.rate = rate
        self.capacity = capacity or rate * 2
        self.per_key = per_key
        self._lock = threading.Lock()
        self._tokens = capacity or rate * 2
        self._last = time.monotonic()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last
        self._tokens = min(self.capacity,
            self._tokens + elapsed * self.rate)
        self._last = now

    def acquire(self, tokens=1.0, wait=False, timeout=5.0):
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            if not wait: return False
            sleep_time = min(
                (tokens - self._tokens) / self.rate, timeout)
        time.sleep(sleep_time)
        return self.acquire(tokens, wait=False)

# Usage:
limiter = TokenBucketRateLimiter(rate=10, capacity=20)
if limiter.acquire(): api.call()  # within limit
limiter.acquire(wait=True)  # block until available
metrics = limiter.metrics  # tokens, rejected, total_wait

💡 Drop in your API client wrapper — never worry about 429 errors again. Full version w/ 10 self-tests (async, per-key, raise mode) in today workspace.