throttlepy

Token bucket and sliding window rate limiting for Python. In-memory or Redis-backed. Sync and async.

Install

pip install throttlepy               # in-memory only
pip install throttlepy[redis]        # + distributed Redis backend

Quickstart

from throttlepy import Limiter, MemoryStorage, TokenBucket

limiter = Limiter(
    algorithm=TokenBucket(capacity=10, refill_per_sec=1.0),
    storage=MemoryStorage(),
)

if limiter.try_acquire("user:42"):
    ...  # do the thing

Two algorithms. Two storages. Two API styles (sync + async). Three ergonomics (try_acquire, acquire+exception, with throttle():).

Algorithms

See docs/algorithms.md for the trade-offs.

Async

from throttlepy import AsyncLimiter, AsyncMemoryStorage, TokenBucket

limiter = AsyncLimiter(
    algorithm=TokenBucket(capacity=10, refill_per_sec=1.0),
    storage=AsyncMemoryStorage(),
)

async with limiter.throttle("user:42"):
    ...

Distributed (Redis)

from redis import Redis
from throttlepy import Limiter, TokenBucket
from throttlepy.storage.redis import RedisStorage

limiter = Limiter(
    algorithm=TokenBucket(capacity=100, refill_per_sec=10.0),
    storage=RedisStorage(Redis(host="localhost")),
)

Redis backend uses atomic Lua scripts for both algorithms — safe under concurrent multi-process load.

Decorator

from throttlepy import Limiter, MemoryStorage, TokenBucket, rate_limit

limiter = Limiter(TokenBucket(capacity=5, refill_per_sec=1.0), MemoryStorage())

@rate_limit(limiter, key=lambda user_id: f"user:{user_id}")
def make_request(user_id: str) -> None:
    ...  # RateLimitExceeded is raised if over limit

Works on async functions too (pass an AsyncLimiter).

Testing your own code

Inject a Clock implementation to control time in tests:

class FakeClock:
    def __init__(self, t: float = 0.0) -> None: self.t = t
    def now(self) -> float: return self.t
    def advance(self, seconds: float) -> None: self.t += seconds

limiter = Limiter(TokenBucket(1, 1.0), MemoryStorage(), clock=FakeClock())

Examples

• examples/basic.py — sync token bucket
• examples/decorator_example.py — @rate_limit
• examples/async_example.py — async sliding window

Benchmarks

just bench

Runs benchmarks/bench.py against MemoryStorage. Report numbers from your own machine — don't quote mine.

Contributing

just setup    # install deps into .venv
just test     # run pytest
just lint     # ruff + mypy
just format   # apply fixes

License

MIT