Quart-Rate-Limiter

Quart-Rate-Limiter is an extension for Quart to allow for rate limits to be defined and enforced on a per route basis and for WebSocket connections. The 429 error response includes a RFC7231 compliant Retry-After header and the successful responses contain headers compliant with the RateLimit Header Fields for HTTP RFC draft.

Quickstart

To add a rate limit first initialise the RateLimiting extension with the application, and then rate limit the route,

app = Quart(__name__)
rate_limiter = RateLimiter(app)

@app.get('/')
@rate_limit(1, timedelta(seconds=10))
async def handler():
    ...

Simple examples

To limit a route to 1 request per second and a maximum of 20 per minute,

@app.route('/')
@rate_limit(1, timedelta(seconds=1))
@rate_limit(20, timedelta(minutes=1))
async def handler():
    ...

Alternatively the limits argument can be used for multiple limits,

@app.route('/')
@rate_limit(
    limits=[
        RateLimit(1, timedelta(seconds=1)),
        RateLimit(20, timedelta(minutes=1)),
    ],
)
async def handler():
    ...

To identify remote users based on their authentication ID, rather than their IP,

async def key_function():
    return current_user.id

RateLimiter(app, key_function=key_function)

The key_function is a coroutine function to allow session lookups if appropriate.

Limiting by response status

A route limit is spent by every request that reaches the route, however that request turns out. It cannot see a request that matched no route, as there is no view function to decorate, and it charges a refused request no more than a successful one. To limit on the outcome instead,

from quart_rate_limiter import StatusLimit

RateLimiter(
    app,
    status_limits=[
        StatusLimit("not_found", (404, 405), [RateLimit(20, timedelta(minutes=1))]),
        StatusLimit("auth", (401, 403), [RateLimit(5, timedelta(minutes=1))]),
        StatusLimit("server_error", range(500, 600), [RateLimit(10, timedelta(minutes=1))]),
    ],
)

These share a single bucket per client rather than one per endpoint, as a budget keyed by path would never be repeated by an agent that invents a new URL each time. status_codes also takes a predicate, and deny_early=True moves the check ahead of the view for cases where running it is itself the attack, such as verifying a password.

WebSocket Rate Limiting

Quart-Rate-Limiter also supports rate limiting for WebSocket connections, similar to fastapi-limiter's WebSocketRateLimiter:

from quart_rate_limiter import RateLimiter, WebSocketRateLimiter, WebSocketRateLimitException
from quart_rate_limiter.redis_store import RedisStore

# Configure global rate limiter with Redis
redis_store = RedisStore("redis://localhost:6379/0")
RateLimiter(app, store=redis_store)

@app.websocket('/ws')
async def websocket_endpoint():
    # Automatically uses the same Redis store as the global RateLimiter
    ratelimit = WebSocketRateLimiter(times=1, seconds=5)

    await websocket.accept()
    while True:
        try:
            data = await websocket.receive()
            await ratelimit(websocket, context_key=data)  # context_key is optional
            await websocket.send(f"Hello, world! You sent: {data}")
        except WebSocketRateLimitException:
            await websocket.send("Rate limited! Please slow down.")

WebSocket rate limiting automatically uses the same storage backend as your global RateLimiter configuration, ensuring consistency across HTTP and WebSocket rate limiting. It supports all the same storage backends (Memory, Redis, Valkey) and key functions as HTTP route rate limiting.

Contributing

Quart-Rate-Limiter is developed on GitHub. You are very welcome to open issues or propose merge requests.

Testing

The best way to test Quart-Rate-Limiter is with Tox,

$ pip install tox
$ tox

this will check the code style and run the tests.

Help

The Quart-Rate-Limiter documentation is the best places to start, after that try searching stack overflow or ask for help on gitter. If you still can't find an answer please open an issue.