⚡ fastapi-sliding-window

fastapi-sliding-window — sliding window rate limiter for FastAPI

(─‿‿─)

    ____           __              _ 
   / __/___ ______/ /_____ _____  (_)
  / /_/ __ `/ ___/ __/ __ `/ __ \/ / 
 / __/ /_/ (__  ) /_/ /_/ / /_/ / /  
/_/  \__,_/____/\__/\__,_/ .___/_/   
                        /_/          
         ___     ___                         _           __             
   _____/ (_)___/ (_)___  ____ _   _      __(_)___  ____/ /___ _      __
  / ___/ / / __  / / __ \/ __ `/  | | /| / / / __ \/ __  / __ \ | /| / /
 (__  ) / / /_/ / / / / / /_/ /   | |/ |/ / / / / / /_/ / /_/ / |/ |/ / 
/____/_/_/\__,_/_/_/ /_/\__, /    |__/|__/_/_/ /_/\__,_/\____/|__/|__/  
                       /____/                                           

📦 installation

pip install fastapi-sliding-window

📑 quick start

from fastapi import FastAPI, Depends
from fastapi_sliding_window import RateLimit

app = FastAPI()

@app.get("/login", dependencies=[Depends(RateLimit(requests=5, window_seconds=60))])
async def login():
    return {"status": "ok"}

🧩 features

• 🎯 3 algorithms — Sliding Window Log (exact), Sliding Window Counter (O(1) memory), Fixed Window
• 💾 in-memory — no Redis required, zero external dependencies
• 🔧 two usage styles — Depends(RateLimit(...)) per-route or RateLimitMiddleware globally
• 📝 standard headers — X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset, Retry-After
• ✅ fully typed — py.typed marker included
• 🚀 async-native — built on asyncio.Lock for thread safety

📖 usage

Depends (per-route)

from fastapi import Depends
from fastapi_sliding_window import RateLimit, Algorithm

# Sliding Window Log (default, most accurate)
@app.get("/api/data", dependencies=[
    Depends(RateLimit(requests=100, window_seconds=60))
])
async def get_data():
    return {"data": "..."}

# Fixed Window (fastest, O(1) memory)
@app.post("/api/upload", dependencies=[
    Depends(RateLimit(requests=10, window_seconds=60, algorithm=Algorithm.FIXED_WINDOW))
])
async def upload():
    return {"status": "uploaded"}

# custom key function (per-user instead of per-IP)
async def user_key(request):
    return request.headers.get("X-User-ID", "anonymous")

@app.get("/api/profile", dependencies=[
    Depends(RateLimit(requests=50, window_seconds=60, key_func=user_key))
])
async def profile():
    return {"user": "..."}

Middleware (global)

from fastapi_sliding_window import RateLimitMiddleware, Algorithm

app.add_middleware(
    RateLimitMiddleware,
    requests=100,
    window_seconds=60.0,
    algorithm=Algorithm.SLIDING_WINDOW_LOG,
    exclude_paths=["/health"],  # skip rate limiting for these paths
)

📊 algorithms

Sliding Window Log

Stores a timestamp for every request. Prunes entries outside the window on each check. Perfectly accurate, memory grows linearly with request volume within the window.

Sliding Window Counter

Maintains two counters (current and previous window) and computes a weighted average. Near-accurate with constant memory. Best balance of accuracy and performance.

Fixed Window

Simple counter per time bucket. Suffers from boundary burst (clients can send 2x the limit across a window boundary).

📝 HTTP headers

successful responses include:

🔗 API Reference

RateLimit(requests, window_seconds, algorithm, key_func, include_headers)

FastAPI dependency for per-route rate limiting.

RateLimitMiddleware(requests, window_seconds, algorithm, key_func, include_headers, exclude_paths)

Starlette middleware for global rate limiting.

📜 license

MIT