limitkit 0.1.0

A small, flexible rate-limiting library for Python: token bucket, fixed/sliding window, leaky bucket. Sync + async. In-memory + Redis.

limitkit

A small, flexible rate-limiting library for Python.

• 4 algorithms: token bucket, fixed window, sliding window, leaky bucket
• 2 storage backends: in-memory (default) and Redis (for cross-process / distributed limiting)
• Per-key limiting: one limiter instance handles per-user / per-IP / per-API-key buckets
• Three API styles: direct check, context manager, decorator
• Sync and async: every API works in both worlds

Install

pip install limitkit             # in-memory only
pip install limitkit[redis]      # adds the Redis backend

60-second tour

from limitkit import TokenBucket

# 5 req/sec sustained, burst tolerance of 5
limiter = TokenBucket(capacity=5, refill_rate=5)

# direct check
if limiter.allow("user:alice"):
    serve_request()

# context manager (raises RateLimitExceeded when over the limit)
with limiter:
    serve_request()

# decorator
@limiter
def search(query):
    ...

Algorithms

All four classes share the same surface (.allow(), with, @, async with, async @). Pick by the behavior you want:

Class	Behavior	When to use
TokenBucket(capacity, refill_rate)	Allows bursts up to capacity; refills at refill_rate per second.	Most APIs. "Up to 100 req with sustained 10 req/sec."
FixedWindow(limit, window)	limit requests allowed per window seconds; resets at window boundaries.	Simple counters: "max 1000 calls per hour." Cheap.
SlidingWindow(limit, window)	Same limit but no boundary burst — the window slides with the request log.	When fixed-window boundary bursts matter.
LeakyBucket(capacity, leak_rate)	Smooths output to a constant rate; queues up to capacity units.	Enforcing a maximum sustained outflow.

Per-key limiting

Pass a key to .allow() and one instance handles many independent buckets:

limiter = TokenBucket(capacity=10, refill_rate=2)

limiter.allow("user:alice")      # alice's bucket
limiter.allow("user:bob")        # bob's bucket — independent
limiter.allow("ip:192.0.2.1")    # any string works
limiter.allow()                  # uses key="default"

Async support

Every algorithm works in async code:

import asyncio
from limitkit import TokenBucket, RateLimitExceeded

limiter = TokenBucket(capacity=5, refill_rate=5)

@limiter
async def fetch(url):
    ...

async def main():
    async with limiter:
        await fetch("...")

    try:
        await fetch("...")
    except RateLimitExceeded:
        ...

asyncio.run(main())

Redis backend (cross-process / distributed)

In-memory limiters protect a single process. With multiple workers behind a load balancer you need shared state — pass a redis_url:

from limitkit import TokenBucket

limiter = TokenBucket(
    capacity=100,
    refill_rate=10,
    redis_url="redis://localhost:6379",
)

limiter.allow("user:alice")   # state lives in Redis; all workers see the same counter

Under the hood every check is a single Lua script call, so updates are atomic — no races between workers. Keys auto-expire after the bucket is fully refilled so abandoned keys don't accumulate.

Heavy / weighted requests

Use n= to mark a call as costing more than 1:

limiter.allow("user:alice", n=5)   # consumes 5 tokens instead of 1

Catching the limit

The decorator and context manager raise RateLimitExceeded:

from limitkit import RateLimitExceeded

try:
    with limiter:
        ...
except RateLimitExceeded:
    return 429

.allow() returns bool instead — pick whichever style fits your code.

License

MIT