Rush: Throttling Algorithms

2021.4.0 · maintenance · verified Sat Apr 11

Rush is a Python library that provides a composable and extensible framework for implementing various rate limiting algorithms and storage backends. It includes a periodic interval rate limiter, a leaky bucket rate limiter (Generic Cell Ratelimiting Algorithm - GCRA), and supports Redis and in-memory dictionary storage. The library emphasizes type annotations and requires Python 3.6 or newer. The current version is 2021.4.0, released in April 2021.

Warnings

gotcha The library's API is explicitly described as 'preliminary' and 'experimental'. While current imports are promised not to break, 'porcelain' (higher-level abstractions) may be added in future versions, potentially altering the recommended usage patterns.
Fix: Refer to the official documentation for the latest recommended API usage and be prepared for potential changes to higher-level abstractions in future updates.
gotcha The last release (2021.4.0) was in April 2021. While the project is marked as 'Production/Stable', development activity appears to be infrequent. Users seeking actively developed or frequently updated libraries might need to consider this pace.
Fix: Evaluate if the existing feature set meets your needs and be aware that new features or rapid bug fixes may not be a priority for the maintainers.

Install

pip install rush Install with pip

Imports

quota
```
from rush import quota
```
throttle
```
from rush import throttle
```
PeriodicLimiter
```
from rush.limiters import periodic
```
DictionaryStore
```
from rush.stores import dictionary
```

Quickstart

This quickstart demonstrates how to set up a basic periodic rate limiter using an in-memory dictionary store, defining a quota of 5000 requests per hour with a burst of 500. It then checks if an operation is limited and prints the result, remaining allowance, and reset time.

from rush import quota
from rush import throttle
from rush.limiters import periodic
from rush.stores import dictionary

t = throttle.Throttle(
    limiter=periodic.PeriodicLimiter(
        store=dictionary.DictionaryStore()
    ),
    rate=quota.Quota.per_hour(
        count=5000,
        burst=500,
    ),
)

limit_result = t.check('expensive-operation/user@example.com', 1)

print(f"Limited: {limit_result.limited}")
print(f"Remaining: {limit_result.remaining}")
print(f"Reset after: {limit_result.reset_after}")

view raw JSON →