FastAPI Sluice

Redis-backed, purely async rate limiting for FastAPI.

FastAPI Sluice is designed for modern async FastAPI applications, providing production-ready Redis-backed rate limiting with interchangeable algorithms and atomic Lua execution.

Features

• ⚡️ 100% async - built for FastAPI's async request lifecycle.
• 🌍 Global or per-route limits — protect your entire API or individual endpoints with the same API.
• 🔒 Atomic Redis operations — Every rate-limiting decision executes inside Redis using Lua, guaranteeing atomic updates under concurrent load without race conditions.
• 🔄 Three Interchangeable algorithms - Choose the algorithm that best matches your traffic profile and resource constraints. Sluice ships out of the box with the three widely used rate-limiting algorithms: Fixed window, sliding window log, and token bucket, all swappable behind a single RateLimiter API.
• 🧩 Flexible identity — Limit by IP, API key, or any other custom attribute. You can also rate limit per-route or globally across your entire API.

Requirements

Installation

Using uv (recommended):

uv add fastapi-sluice

Using pip:

pip install fastapi-sluice

Using Poetry:

poetry add fastapi-sluice

Usage

Start by connecting to a Redis server and creating a RateLimiter instance:

from redis.asyncio import Redis
from fastapi_sluice import RateLimiter, FixedWindow, SlidingWindow, TokenBucket

redis = Redis(host="localhost", port=6379)
limiter = RateLimiter(redis=redis)

Per-route limiting

Apply a limit to a specific route by passing limiter.limit() as a dependency. Each route gets its own counter, keyed by IP address by default. You can pass a scope string to group requests to use the same counter. By default, the request path is used. Use scope when you want multiple routes or parameterized URLs to share the same counter.

from fastapi import FastAPI, Depends
from fastapi_sluice import FixedWindow

app = FastAPI()

@app.get("/items")
async def get_items(_=Depends(limiter.limit(algorithm=FixedWindow(limit=5, window_seconds=60), scope="items"))):
    return {"items": []}

Global limiting

Use RateLimitMiddleware to enforce an API-wide cap that applies to every route. A single counter per identity is shared across all endpoints:

from fastapi_sluice import RateLimitMiddleware, SlidingWindow

app.add_middleware(
    RateLimitMiddleware,
    limiter=limiter,
    algorithm=SlidingWindow(limit=500, window_seconds=60),
)

Choosing an algorithm

Each algorithm suits a different traffic profile:

from fastapi_sluice import FixedWindow, SlidingWindow, TokenBucket

# Fixed window — simplest, cheapest. Best for low-stakes limits.
FixedWindow(limit=100, window_seconds=60)

# Sliding window — accurate per-second fairness, no boundary bursts.
SlidingWindow(limit=100, window_seconds=60)

# Token bucket — sustain a rate while allowing controlled bursts.
TokenBucket(capacity=50, refill_rate=10)  # 10 req/s, burst up to 50

Rate limit responses

When a client exceeds the limit, Sluice returns a 429 Too Many Requests response with the following headers:

Example response:

HTTP/1.1 429 Too Many Requests
Retry-After: 30
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0

{"detail": "Too many requests"}

Custom Client Identifier

You can override the default IP-based key with any attribute like API key or user ID. The callable must return a unique string per identity since this value becomes the rate limit bucket key in Redis. Just create a sync or async callable:

from fastapi import Request

def get_api_key(request: Request) -> str:
    api_key = request.headers.get("X-API-Key")
    if api_key is None:
        raise ValueError("X-API-Key header is missing")
    return api_key

@app.get("/items")
async def get_items(_=Depends(limiter.limit(algorithm=FixedWindow(limit=30, window_seconds=60), key_func=get_api_key))):
    return {"items": []}