PyrateLimiter
PyrateLimiter is a Python library implementing the Leaky-Bucket Algorithm for rate limiting. It supports both synchronous and asynchronous workflows and offers various backends like in-memory, SQLite, Redis, and PostgreSQL for persistent limit tracking. Currently at version 4.1.0, it maintains an active development and release cadence, with several minor and patch releases in the last year.
Warnings
- breaking The `Limiter` constructor API changed significantly in v4.0.0. It now expects a single `Bucket` instance or a `BucketFactory` instead of a list of `Rate` objects directly. The `RequestRate` class was also renamed to `Rate`.
- breaking The `max_delay` argument for configuring rate limit delays in the `Limiter` constructor was removed in v4.0.0. Delay behavior is now integrated or configured differently at the bucket level.
- gotcha For Redis or PostgreSQL backends, you must install the optional dependencies (e.g., `pyrate-limiter[redis]` or `pyrate-limiter[postgres]`). Without them, these backends will not be available.
- gotcha When defining multiple rates for a single limiter, they must be ordered correctly: by interval length (least to greatest) and by the ratio of limit/interval (greatest to least). Incorrect ordering can lead to unexpected rate limiting behavior.
- gotcha The default `InMemoryBucket` does not share state across multiple processes or distributed environments. For multiprocessing or distributed applications, use persistent backends like `SQLiteBucket` (with `use_file_lock=True`) or `RedisBucket`/`PostgresBucket`.
Install
-
pip install pyrate-limiter -
pip install pyrate-limiter[redis] -
pip install pyrate-limiter[postgres] -
pip install pyrate-limiter[all]
Imports
- Limiter
from pyrate_limiter import Limiter
- Rate
from pyrate_limiter import Duration, Rate
Quickstart
from pyrate_limiter import Duration, Rate, Limiter
# Limit 5 requests within 2 seconds
limiter = Limiter(Rate(5, Duration.SECOND * 2))
# Blocking mode (default) - waits until permit available
for i in range(6):
limiter.try_acquire(str(i))
print(f"Acquired permit {i}")
# Non-blocking mode - returns False if bucket full
for i in range(6):
success = limiter.try_acquire(str(i), blocking=False)
if not success:
print(f"Rate limited at {i}")