Rate Limiting in FastAPI with SlowAPI

Why Rate Limiting Matters

Without rate limiting, a single client can overwhelm your API with requests — whether through a bug, a scraper, or a deliberate attack. Rate limiting protects your server resources, ensures fair usage across clients, and prevents abuse.

SlowAPI is a rate limiting library for ASGI frameworks, built on top of the popular limits library. It integrates cleanly with FastAPI.

Installation

pip install slowapi

For Redis-backed storage in production:

pip install slowapi redis

Basic Setup

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# Create limiter with client identification function
limiter = Limiter(key_func=get_remote_address)

app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/public")
@limiter.limit("10/minute")
async def public_endpoint(request: Request):
    return {"message": "This endpoint allows 10 requests per minute"}

@app.get("/search")
@limiter.limit("5/minute")
async def search(request: Request, q: str):
    return {"results": [], "query": q}

The key_func determines how to identify each client. get_remote_address uses the client’s IP address. The @limiter.limit() decorator sets the rate for each endpoint. When the limit is exceeded, SlowAPI returns a 429 Too Many Requests response.

Rate Limit String Formats

SlowAPI supports flexible rate expressions.

@limiter.limit("100/hour")         # 100 requests per hour
@limiter.limit("10/minute")        # 10 requests per minute
@limiter.limit("1/second")         # 1 request per second
@limiter.limit("5000/day")         # 5000 requests per day
@limiter.limit("5/minute;100/hour") # Multiple limits (both apply)

When multiple limits are specified with a semicolon, all limits are enforced simultaneously. A request is rejected if any of the limits are exceeded.

Custom Key Functions

Rate limiting by IP works for basic cases, but you often want to limit by authenticated user, API key, or a combination.

from fastapi import Header, Depends

def get_api_key(request: Request) -> str:
    """Rate limit by API key from header."""
    api_key = request.headers.get("X-API-Key", "anonymous")
    return api_key

def get_user_or_ip(request: Request) -> str:
    """Rate limit by user ID if authenticated, otherwise by IP."""
    user_id = request.headers.get("X-User-ID")
    if user_id:
        return f"user:{user_id}"
    return f"ip:{get_remote_address(request)}"

# Use custom key function for specific endpoints
api_key_limiter = Limiter(key_func=get_api_key)

@app.get("/api/data")
@limiter.limit("30/minute", key_func=get_user_or_ip)
async def get_data(request: Request):
    return {"data": "rate limited by user or IP"}

Different Limits for Different User Tiers

Apply dynamic rate limits based on the user’s subscription tier.

def get_rate_limit_by_tier(request: Request) -> str:
    """Return different rate limits based on user tier."""
    api_key = request.headers.get("X-API-Key", "")
    tier = get_user_tier(api_key)  # Your lookup function

    limits = {
        "free": "10/minute",
        "basic": "60/minute",
        "pro": "300/minute",
        "enterprise": "1000/minute"
    }
    return limits.get(tier, "5/minute")

@app.get("/api/process")
@limiter.limit(limit_value=get_rate_limit_by_tier)
async def process_data(request: Request):
    return {"status": "processed"}

Passing a callable to limit_value lets you compute the rate limit dynamically per request.

Global Rate Limits

Apply a default rate limit to all endpoints, then override specific ones.

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["200/day", "50/hour"]  # Apply to all routes
)

app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# This endpoint uses the global default (200/day, 50/hour)
@app.get("/standard")
async def standard_endpoint(request: Request):
    return {"message": "Using global limits"}

# This endpoint overrides with a stricter limit
@app.get("/expensive")
@limiter.limit("5/minute")
async def expensive_endpoint(request: Request):
    return {"message": "Expensive operation, stricter limit"}

# Exempt this endpoint from rate limiting
@app.get("/health")
@limiter.exempt
async def health_check(request: Request):
    return {"status": "healthy"}

The @limiter.exempt decorator removes rate limiting from endpoints that should always be accessible, like health checks.

Redis-Backed Storage for Production

In-memory storage resets when the server restarts and does not work across multiple server instances. Use Redis for persistent, distributed rate limiting.

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://localhost:6379/0",
    default_limits=["200/day", "50/hour"]
)

That is all you need. SlowAPI handles the Redis connection and stores rate limit counters there. All server instances share the same counters, giving you consistent rate limiting across your fleet.

Redis with Authentication

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="redis://:your-password@redis-host:6379/0",
    default_limits=["100/hour"]
)

Custom Error Responses

The default 429 response is plain. Customize it with useful information for API consumers.

from fastapi.responses import JSONResponse
from slowapi.errors import RateLimitExceeded

def custom_rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "error": "rate_limit_exceeded",
            "message": f"Rate limit exceeded: {exc.detail}",
            "retry_after": exc.detail,
            "docs": "https://api.example.com/docs/rate-limits"
        },
        headers={
            "Retry-After": str(60),
            "X-RateLimit-Limit": str(exc.detail)
        }
    )

app.add_exception_handler(RateLimitExceeded, custom_rate_limit_handler)

Adding Rate Limit Headers

Let clients know their current rate limit status by adding headers to every response.

from fastapi import Response

@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
    response = await call_next(request)

    # SlowAPI automatically adds these headers when configured:
    # X-RateLimit-Limit: the rate limit ceiling
    # X-RateLimit-Remaining: requests left in the window
    # X-RateLimit-Reset: UTC epoch seconds when the window resets

    return response

SlowAPI adds X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset headers automatically when rate limits are applied.

Rate Limiting by Endpoint Groups

Group related endpoints under shared rate limits using shared keys.

def make_shared_key(group: str):
    def key_func(request: Request) -> str:
        client = get_remote_address(request)
        return f"{group}:{client}"
    return key_func

# These three endpoints share a single rate limit pool
write_key = make_shared_key("write_ops")

@app.post("/items")
@limiter.limit("20/minute", key_func=write_key)
async def create_item(request: Request):
    return {"created": True}

@app.put("/items/{item_id}")
@limiter.limit("20/minute", key_func=write_key)
async def update_item(request: Request, item_id: int):
    return {"updated": item_id}

@app.delete("/items/{item_id}")
@limiter.limit("20/minute", key_func=write_key)
async def delete_item(request: Request, item_id: int):
    return {"deleted": item_id}

All three write endpoints share the same counter. A client making 10 creates and 10 updates will hit the limit, even though they did not exceed any single endpoint’s quota.

Testing Rate Limits

from fastapi.testclient import TestClient

def test_rate_limit_enforcement():
    client = TestClient(app)

    # Send requests up to the limit
    for i in range(10):
        response = client.get("/public")
        assert response.status_code == 200

    # The next request should be rejected
    response = client.get("/public")
    assert response.status_code == 429

def test_health_check_exempt():
    client = TestClient(app)

    # Health check should never be rate limited
    for _ in range(500):
        response = client.get("/health")
        assert response.status_code == 200

Key Takeaways

SlowAPI makes rate limiting in FastAPI straightforward. Start with IP-based limiting using get_remote_address and simple rate strings. Use custom key functions for API-key or user-based limits. Apply global defaults and override per-endpoint as needed. In production, back your limiter with Redis for persistence and multi-instance consistency. Always exempt health check endpoints and provide informative error responses so API consumers know when and how to retry.