ratekeeper-toolkit 0.1.0

A scalable, pluggable rate limiting toolkit (Token Bucket & Sliding Window) with in-memory and Redis-backed distributed support, plus FastAPI middleware.

ratekeeper_toolkit

A small, scalable rate limiting toolkit for Python. Protects an API, a function, or a whole app from being overloaded or abused — the same job rate limiters do at OpenAI, GitHub, and Stripe, sized for a single service instead of a whole platform.

• Two algorithms: Token Bucket (allows short bursts) and Sliding Window (strict, no double-burst at window edges).
• Two backends: MemoryBackend for a single process (thread-safe, sharded locks so it doesn't bottleneck under load), and RedisBackend for multiple processes/hosts sharing one limit (atomic Lua scripts — no race conditions between instances).
• Three ways to use it: plain function calls, a decorator, or FastAPI/ Starlette middleware.
• Sync and async APIs throughout.

Install

pip install ratekeeper_toolkit

# with Redis support (multi-process / multi-host deployments)
pip install ratekeeper_toolkit[redis]

# with FastAPI middleware support
pip install ratekeeper_toolkit[fastapi]

Quickstart

from ratekeeper_toolkit import create_limiter

# 100 requests/minute per key, refilling smoothly (100/60 ≈ 1.67 tokens/sec)
limiter = create_limiter("token_bucket", capacity=100, refill_rate=1.67)

if limiter.allow(user_id):
    process_request()
else:
    return "429 Too Many Requests"

That's the whole API for the simple case. Everything else below is the same idea applied to specific situations.

The two algorithms

Token Bucket — a bucket holds up to capacity tokens and refills at refill_rate tokens/second. Good default: lets a client burst up to capacity requests instantly, then settles into the steady rate.

from ratekeeper_toolkit import create_limiter

limiter = create_limiter("token_bucket", capacity=20, refill_rate=5)  # burst of 20, then 5/sec

Sliding Window — allows at most limit requests in any rolling window_seconds window. No burst allowance; stricter and more predictable.

limiter = create_limiter("sliding_window", limit=5, window_seconds=60)  # 5 requests per rolling minute

Both return a RateLimitResult:

result = limiter.check(user_id)
result.allowed       # bool
result.remaining     # tokens or requests left
result.limit         # configured capacity/limit
result.retry_after   # seconds until you can retry, if denied

Choosing a backend

In-process app / single container — use the default MemoryBackend, nothing to configure:

from ratekeeper_toolkit import create_limiter

limiter = create_limiter("token_bucket", capacity=100, refill_rate=10)

Multiple app instances behind a load balancer — a MemoryBackend per instance means each instance has its own separate limit, so the effective limit multiplies by instance count. Use RedisBackend so every instance enforces the same shared limit:

from ratekeeper_toolkit import create_limiter, RedisBackend

limiter = create_limiter(
    "token_bucket",
    capacity=100,
    refill_rate=10,
    backend=RedisBackend(url="redis://localhost:6379/0"),
)

Usage patterns

1. Direct check, before an expensive call

The most common integration — guard anything that costs money or load before you do it (an LLM call, a third-party API, a login attempt):

from ratekeeper_toolkit import create_limiter

limiter = create_limiter("token_bucket", capacity=100, refill_rate=1.67)  # 100/min per user

def call_openai(user_id, prompt):
    if not limiter.allow(user_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    return openai_client.chat.completions.create(...)

Async version:

if not await limiter.aallow(user_id):
    raise HTTPException(status_code=429)

2. Decorator

Enforce a limit on any function — sync or async — without touching its body:

from ratekeeper_toolkit import create_limiter, rate_limited, RateLimitExceeded

limiter = create_limiter("sliding_window", limit=5, window_seconds=600)  # 5 OTPs / 10 min

@rate_limited(limiter, key=lambda user_id: user_id)
def send_otp(user_id):
    sms_client.send(user_id, generate_otp())

try:
    send_otp("user-42")
except RateLimitExceeded as e:
    print(f"blocked, retry after {e.retry_after:.0f}s")

Works the same way for async def functions.

3. FastAPI / Starlette middleware

Protect every route in an app with no per-endpoint code:

from fastapi import FastAPI
from ratekeeper_toolkit import RateLimitMiddleware, create_limiter, RedisBackend

app = FastAPI()

app.add_middleware(
    RateLimitMiddleware,
    limiter=create_limiter(
        "token_bucket", capacity=100, refill_rate=1.67,
        backend=RedisBackend(url="redis://localhost:6379/0"),
    ),
    key_func=lambda request: request.headers.get("x-api-key", request.client.host),
)

Denied requests get a 429 with X-RateLimit-* and Retry-After headers automatically. Allowed requests get X-RateLimit-* headers too, so clients can see how close they are to the limit.

Real-world examples this maps to

• Per-plan API limits (Free/Basic/Premium tiers) — one token_bucket limiter, key_func reads the plan's capacity/refill_rate per customer.
• Login brute-force protection — sliding_window, limit=5, window_seconds=60, keyed by username or IP.
• OTP abuse prevention — sliding_window, limit=3, window_seconds=600, keyed by phone number.
• Protecting a paid upstream API (OpenAI, an internal Jira/GitLab API, etc.) — a token_bucket sized to the upstream's own published limit, so your service degrades gracefully instead of getting throttled upstream.
• Multi-tenant SaaS — one limiter instance per tenant tier, backed by RedisBackend so the limit holds across all your app instances.

API reference (short version)

create_limiter(algorithm, backend=None, **config) -> TokenBucketLimiter | SlidingWindowLimiter

limiter.check(key, cost=1.0) -> RateLimitResult      # sync, never raises
limiter.acheck(key, cost=1.0) -> RateLimitResult      # async, never raises
limiter.allow(key, cost=1.0) -> bool
limiter.aallow(key, cost=1.0) -> bool
limiter.enforce(key, cost=1.0) -> RateLimitResult     # raises RateLimitExceeded if denied
limiter.aenforce(key, cost=1.0) -> RateLimitResult

rate_limited(limiter, key=None, cost=1.0)             # decorator, sync + async functions

MemoryBackend(shards=64)                              # single-process, thread-safe
RedisBackend(url=..., key_prefix="ratekeeper_toolkit:", ttl_seconds=86400)  # multi-process, atomic

RateLimitMiddleware(app, limiter, key_func=None, cost=1.0)  # FastAPI/Starlette

Development

git clone <your fork>
cd ratekeeper_toolkit
pip install -e ".[dev]"
pytest tests/ -v
python loadtest/load_test.py --concurrency 200 --requests 20000

See PUBLISHING.md if you're maintaining this package and need to cut a release.

License

MIT