Comment ajouter un rate limiter à notre application FastAPI avec redis

 

Dans ce tutoriel, on met en place un rate limiting pour une API FastAPI à l’aide de la bibliothèque fastapi-limiter, avec Redis. L’objectif est de protéger vos endpoints contre les abus (pics de trafic, scripts, DDoS applicatif léger) et de mieux contrôler votre consommation de ressources.

Prérequis : savoir démarrer une API minimaliste. Voir Python : Comment faire une api web avec FastAPI.

Installation

Installez les dépendances nécessaires :

pip install fastapi uvicorn "fastapi-limiter==0.1.6" redis

Sous Windows (PowerShell), vous pouvez faire :

python -m pip install fastapi uvicorn "fastapi-limiter==0.1.6" redis

Note : ce tutoriel épingle fastapi-limiter en version 0.1.6. La version 0.2.0 a introduit des changements d’API (refonte interne du limiteur) qui rendent certains exemples ci-dessous incompatibles. Épingler la version garantit que le code fonctionne tel quel.

Démarrer un Redis local (docker-compose)

On utilise docker pour déployer un serveur Redis rapidement en local :

# docker-compose.yml
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: [ "redis-server", "--appendonly", "yes" ]

Lancez :

docker compose up -d

Mise en place minimale avec fastapi-limiter

fastapi-limiter s’initialise au démarrage de l’application avec un client Redis. Ensuite, on ajoute une dépendance RateLimiter sur les routes à protéger.

Mise en place du limiter

On initialise le rate limiter dans le gestionnaire de contexte lifespan utilisé par FastAPI :

# app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import redis.asyncio as redis

redis_client: redis.Redis | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    # Connexion Redis (adapter l'URL si besoin: auth, DB, TLS, etc.)
    redis_client = redis.from_url(
        "redis://localhost:6379", encoding="utf-8", decode_responses=True
    )
    await FastAPILimiter.init(redis_client, prefix="fastapi-limiter")
    yield
    # Fermeture propre
    assert redis_client is not None
    await redis_client.aclose()

app = FastAPI(lifespan=lifespan)

Note : Vous pouvez réutiliser le même client redis que pour votre cache

Puis, on ajoute à nos routes notre limiteur :

# Cette route autorise 5 requêtes par minute (par identifiant; voir plus bas)
@app.get("/ping", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
async def ping():
    return {"pong": True}

Démarrage :

uvicorn app:app --reload

Test rapide :

# Faites plus de 5 appels en moins de 60s pour observer l'erreur 429
for i in {1..7}; do curl -i http://127.0.0.1:8000/ping; echo; done

Par défaut, fastapi-limiter identifie le client par son IP (en s’aidant des en-têtes classiques si vous utilisez un proxy). Vous pouvez cependant personnaliser cet identifiant si besoin.


Personnaliser l’identifiant (IP, clé API, utilisateur, etc.)

Souvent, on veut limiter par clé API ou par utilisateur authentifié plutôt que par IP. Pour cela, on peut fournir une fonction identifier au RateLimiter.

fastapi-limiter appelle l’identifier avec await : il doit donc s’agir d’une fonction asynchrone (async def), et non d’un lambda synchrone.

from fastapi import Request

# Limite par clé API (X-API-Key) si présente, sinon par IP
async def api_key_identifier(request: Request) -> str:
    return request.headers.get("X-API-Key") or request.client.host

# Limite 100 requêtes par 24h et par clé API (X-API-Key), sinon par IP
@app.get(
    "/data",
    dependencies=[
        Depends(
            RateLimiter(
                times=100,
                hours=24,
                identifier=api_key_identifier,
            )
        )
    ],
)
async def get_data(request: Request):
    return {"ok": True}

Autre exemple : limitation par utilisateur connecté (ex: request.state.user.id ou request.user.id selon votre middleware d’authentification).

async def user_identifier(request: Request) -> str:
    # Identifiant de l'utilisateur connecté si disponible, sinon IP
    user = getattr(request, "user", None)
    return str(getattr(user, "id", None) or request.client.host)

@app.get(
    "/me",
    dependencies=[
        Depends(
            RateLimiter(
                times=60,
                minutes=1,
                identifier=user_identifier,
            )
        )
    ],
)
async def me():
    return {"me": True}

Appliquer une limite par défaut à un groupe de routes

Vous pouvez appliquer un rate limit à l’échelle d’un router, afin qu’il s’applique à toutes les routes incluses.

from fastapi import APIRouter

api_router = APIRouter(
    prefix="/api",
    dependencies=[Depends(RateLimiter(times=120, minutes=1))],  # par défaut: 120 req/min
)

@api_router.get("/items")
async def list_items():
    return {"items": []}

@api_router.post("/items")
async def create_item():
    return {"created": True}

app.include_router(api_router)

Vous pouvez toujours surcharger/compléter le comportement sur une route précise en ajoutant un autre Depends(RateLimiter(...)) directement sur l’endpoint.


Cas derrière un reverse proxy (Nginx, Traefik, Cloudflare)

Pour que l’IP réelle du client soit correctement vue, pensez à activer la prise en compte des en-têtes proxy. Par exemple :

from starlette.middleware import Middleware
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

# trusted_hosts : IP ou hôtes des proxies de confiance dont on accepte les en-têtes
# X-Forwarded-* ("*" fait confiance à tous les proxies, à réserver aux environnements
# où l'accès direct à l'application est impossible)
app = FastAPI(lifespan=lifespan, middleware=[Middleware(ProxyHeadersMiddleware, trusted_hosts="*")])

Gérer le 429 Too Many Requests

Quand la limite est dépassée, fastapi-limiter soulève une HTTPException(429). Vous pouvez personnaliser la gestion globale avec un handler FastAPI pour retourner un message JSON cohérent avec votre API.

from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(429)
async def too_many_requests_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=429,
        content={
            "error": "too_many_requests",
            "detail": exc.detail or "Rate limit exceeded",
        },
        headers={"Retry-After": "60"},
    )

Bonnes pratiques et points d’attention

  • Granularité : adaptez les paramètres (secondes, minutes, heures) à vos usages.
  • Identifiant : préférez un identifiant stable (clé API, user id) quand c’est pertinent.
  • Proxies : gérez correctement les IP réelles (ProxyHeadersMiddleware, trusted hops).
  • Endpoints sensibles : combinez avec de l’authentification, voire du captcha sur les routes publiques.
  • Observabilité : loggez les 429 et surveillez vos métriques.

Pour aller plus loin