fastapi-limiter-valkey 0.2.0

A request rate limiter for fastapi, backed by Valkey

fastapi-limiter-valkey

Introduction

FastAPI-Limiter-Valkey is a rate limiting tool for fastapi routes, backed by Valkey.

It is a friendly fork of FastAPI-Limiter, which since 0.2.0 is powered by pyrate-limiter and is fully storage-agnostic. This package adds the missing piece for Valkey users: a ValkeyBucket — a pyrate-limiter bucket typed for the valkey client (the Redis protocol works unchanged, so it runs on a Valkey server) — and re-exports the upstream RateLimiter dependencies so you only need a single install.

Since pyrate-limiter's RedisBucket only imports redis under TYPE_CHECKING, a Valkey client already works at runtime. ValkeyBucket exists so that type checkers are happy and the intent is explicit — you never install redis.

Install

> pip install fastapi-limiter-valkey

Quick Start

Build a Limiter from a ValkeyBucket and pass it to the RateLimiter dependency. Because creating a Valkey-backed bucket is async (it loads a Lua script onto the server), build the limiters in the app's lifespan:

from contextlib import asynccontextmanager

import uvicorn
import valkey.asyncio as valkey
from fastapi import Depends, FastAPI, Request, Response
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter_valkey import RateLimiter, ValkeyBucket

limiters: dict[str, RateLimiter] = {}


@asynccontextmanager
async def lifespan(_: FastAPI):
    client = valkey.from_url("valkey://localhost:6379", encoding="utf-8")
    bucket = await ValkeyBucket.init([Rate(2, Duration.SECOND * 5)], client, "index")
    limiters["index"] = RateLimiter(limiter=Limiter(bucket))
    yield
    await client.aclose()


app = FastAPI(lifespan=lifespan)


def rate_limit(key: str):
    async def dependency(request: Request, response: Response):
        return await limiters[key](request, response)

    return dependency


@app.get("/", dependencies=[Depends(rate_limit("index"))])
async def index():
    return {"msg": "Hello World"}


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)

See examples/main.py for multiple limiters, websockets and the skip decorator.

Usage

ValkeyBucket

ValkeyBucket.init(rates, client, bucket_key) is pyrate_limiter.RedisBucket.init typed for a Valkey client. It accepts both sync and async clients; with an async client it returns an awaitable that resolves to the bucket. Wrap the result in a pyrate_limiter.Limiter and hand that to RateLimiter.

RateLimiter

RateLimiter (re-exported from upstream fastapi-limiter) accepts:

• limiter: a pyrate_limiter.Limiter instance that defines the rate limiting rules.
• identifier: a callable to identify the request source, default is by IP + path.
• callback: a callable invoked when the rate limit is exceeded, default raises HTTPException with 429 status code.
• blocking: whether to block the request when the rate limit is exceeded, default is False.

identifier

Default is ip + path; override it for e.g. userid:

async def default_identifier(request: Union[Request, WebSocket]):
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        ip = forwarded.split(",")[0]
    elif request.client:
        ip = request.client.host
    else:
        ip = "127.0.0.1"
    return ip + ":" + request.scope["path"]

callback

Callback when the rate limit is exceeded, default raises HTTPException with 429:

def default_callback(*args, **kwargs):
    raise HTTPException(
        HTTP_429_TOO_MANY_REQUESTS,
        "Too Many Requests",
    )

Multiple limiters

You can use multiple limiters in one route. Keep the stricter limiter (lower seconds/times ratio) first.

@app.get(
    "/multiple",
    dependencies=[Depends(rate_limit("multiple_short")), Depends(rate_limit("multiple_long"))],
)
async def multiple():
    return {"msg": "Hello World"}

Skip rate limiting

Use the skip_limiter decorator to skip rate limiting for a specific route:

from fastapi_limiter_valkey import skip_limiter

@app.get("/skip", dependencies=[Depends(rate_limit("skip"))])
@skip_limiter
async def skip_route():
    return {"msg": "This route skips rate limiting"}

Rate limiting within a websocket

from fastapi_limiter_valkey import WebSocketRateLimiter

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    ratelimit = ws_limiters["ws"]  # a WebSocketRateLimiter built in the lifespan
    while True:
        try:
            data = await websocket.receive_text()
            await ratelimit(websocket, context_key=data)  # NB: context_key is optional
            await websocket.send_text("Hello, world")
        except HTTPException:
            await websocket.send_text("Hello again")

License

This project is licensed under the Apache-2.0 License.