Rate limit FastAPI endpoints with Redis

Approach

Implement a reusable dependency that increments a Redis counter per key and window atomically (pipeline with INCR + EXPIRE), raising HTTPException(429) with Retry-After. Configure the key function per route (user ID, API key or client IP).

Solution

import time
from collections.abc import Callable
from fastapi import Depends, HTTPException, Request
from redis.asyncio import Redis

redis = Redis.from_url("redis://redis:6379/0")


def rate_limit(limit: int, window: int, key_func: Callable[[Request], str]):
    async def dependency(request: Request) -> None:
        bucket = int(time.time() // window)
        key = f"rl:{request.url.path}:{key_func(request)}:{bucket}"
        async with redis.pipeline(transaction=True) as pipe:
            count, _ = await pipe.incr(key).expire(key, window).execute()
        if count > limit:
            retry_after = window - int(time.time() % window)
            raise HTTPException(429, "Too many requests", headers={"Retry-After": str(retry_after)})
    return Depends(dependency)


def client_ip(request: Request) -> str:
    return request.client.host if request.client else "unknown"   # behind a proxy: run uvicorn with --proxy-headers


@router.post("/auth/token", dependencies=[rate_limit(5, 60, client_ip)])
async def login(...): ...


@router.get("/reports", dependencies=[rate_limit(100, 60, lambda r: r.headers.get("x-api-key", "anon"))])
async def reports(...): ...

Pitfalls

• In-memory limiters (each worker has its own count).
• Using the proxy's IP because proxy headers aren't trusted.
• Failing closed for all traffic when Redis is down — decide per route.

Variations

• slowapi (limits library) offers decorators and multiple strategies.
• Enforce limits at the API gateway (Kong, Envoy, Cloudflare) for coarse protection.

Further reading: Redis patterns