This is an extension for Robyn that allows rate limiting your API.
You can get robyn-rate-limits
from PyPI, which means you can install it with pip easily:
python -m pip install robyn-rate-limits
If you would like to use Redis
as a store, enable the redis feature:
python -m pip install robyn-rate-limits[redis]
Define your API normally as you would with Robyn, and add a limiter middleware:
from robyn import Robyn
from robyn_rate_limits import InMemoryStore
from robyn_rate_limits import RateLimiter
app = Robyn(__file__)
limiter = RateLimiter(store=InMemoryStore, calls_limit=3, limit_ttl=100)
@app.before_request()
def middleware(request: Request):
return limiter.handle_request(app, request)
@app.get("/")
def h():
return "Hello, World!"
app.start(port=8080)
Robyn Rate Limits supports multiple caching strategies to suit different use cases, both in-memory and Redis-based:
from robyn_rate_limits import InMemoryStore
limiter = RateLimiter(store=InMemoryStore, calls_limit=3, limit_ttl=100)
from robyn_rate_limits import InMemoryFixedWindowStore
limiter = RateLimiter(store=InMemoryFixedWindowStore, calls_limit=3, limit_ttl=100, window_size=60)
from robyn_rate_limits import InMemoryTokenBucketStore
limiter = RateLimiter(store=InMemoryTokenBucketStore, calls_limit=3, refill_rate=1, capacity=5)
To use Redis-based stores, first establish a Redis connection:
import redis
redis_conn = redis.Redis(host='localhost', port=6379, db=0)
from robyn_rate_limits import RedisStore
limiter = RateLimiter(store=RedisStore, calls_limit=3, limit_ttl=100, redis=redis_conn)
from robyn_rate_limits import RedisFixedWindowStore
limiter = RateLimiter(store=RedisFixedWindowStore, calls_limit=3, limit_ttl=100, window_size=60, redis=redis_conn)
from robyn_rate_limits import RedisTokenBucketStore
limiter = RateLimiter(store=RedisTokenBucketStore, calls_limit=3, refill_rate=1.0, capacity=5, redis=redis_conn)
Choose the appropriate store based on your application's needs:
- Use the Sliding Window for a balance between accuracy and performance.
- Use Fixed Window for simplicity and when precise timing isn't critical.
- Use Token Bucket when you want to allow short bursts of traffic.
- Use Redis-based stores for distributed systems or when you need persistence.
The extension is designed in a way that you can implement your own store if you would like to use a different store or algorithm.
The identity of the client that the rate is limited by is automatically detected:
-
For endpoints that require authentication, the rate is enforced by token.
-
For endpoints that are open, the rate is enforced by IP.
If you add more stores or algorithms and would like them to be part of the official package you are more than welcomed to!
Please read the contributing guide for the guidelines.
Feel free to open issues if you have any question or suggestion.
-
Install the development dependencies:
poetry install --with dev
-
Install the pre-commit git hooks:
pre-commit install
-
Run
poetry run test_server
. This will run a server containing several examples of routes we use for testing purposes. You can see them attests/base_routes.py
. You can modify or add some to your likings.
You can then request the server you ran from an other terminal. Here is a GET
request done using curl for example:
curl http://localhost:8080/darwin/11/test
# 200
curl http://localhost:8080/darwin/11/test
# 200
curl http://localhost:8080/darwin/11/test
# 200
curl http://localhost:8080/darwin/11/test
# 429