fastapi-sluice 0.1.0

FastAPI rate limiter backed by Redis

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

Dependency	Supported Versions
Python	3.10+
FastAPI	0.100+
Redis server	7.4+
redis-py	4.5+

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

Algorithm	Performance / Memory	Precision & Fairness	Best Used For
Fixed Window	O(1) time and space	Low — susceptible to boundary bursting	Standard API protection where perfection isn't critical
Sliding Window	O(N) space per user, higher memory cost	Highest — accurate across shifting time windows	Critical or expensive endpoints (e.g. AI generation, payment processing)
Token Bucket	O(1) time and space	High — allows controlled traffic bursts	General-purpose API protection and microservices

Rate limit responses

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

Header	Description
Retry-After	Seconds to wait before retrying
X-RateLimit-Limit	Maximum requests allowed in the window
X-RateLimit-Remaining	Requests remaining in the current window

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": []}