fapi-limiter

Flexible rate limiting for FastAPI powered by pyrate-limiter.

This project is being developed with AI assistance (GitHub Copilot), but all code, design decisions, and implementation details are reviewed and validated by the author.

Installation

# pip
pip install fapi-limiter

# uv
uv add fapi-limiter

Usage

Middleware (app-wide)

Apply one limit per client across the whole application.

from fastapi import FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fapi_limiter import (
    DynamicBucketFactory,
    RateLimiterMiddleware,
    setup_inmemory_bucket,
)

app = FastAPI()

limiter = Limiter(
    DynamicBucketFactory(setup_inmemory_bucket([Rate(10, Duration.MINUTE)]))
)
app.add_middleware(RateLimiterMiddleware, limiter=limiter)

Dependency (per-route)

Apply a limit per client and route using FastAPI's dependency injection. It also works on WebSocket routes, where a client over the limit is denied with a 429 before the connection is accepted.

from fastapi import Depends, FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fapi_limiter import DynamicBucketFactory, RateLimiter, setup_inmemory_bucket

app = FastAPI()

limiter = Limiter(
    DynamicBucketFactory(setup_inmemory_bucket([Rate(10, Duration.MINUTE)]))
)


@app.get("/limited", dependencies=[Depends(RateLimiter(limiter))])
def limited():
    return {"ok": True}

SQLite backend (persistent, multi-process)

use_file_lock=True requires the filelock package. All identifiers share one connection and one table, fapi_limiter_items.

from fastapi import FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fapi_limiter import (
    DynamicBucketFactory,
    RateLimiterMiddleware,
    setup_sqlite_bucket,
)

app = FastAPI()

limiter = Limiter(
    DynamicBucketFactory(
        setup_sqlite_bucket(
            [Rate(10, Duration.MINUTE)],
            db_path="rate_limiter.db",
            use_file_lock=True,  # required when sharing across processes
        )
    )
)
app.add_middleware(RateLimiterMiddleware, limiter=limiter)

Redis backend

Requires the redis package. Both redis.Redis and redis.asyncio.Redis clients work.

from fastapi import FastAPI
from pyrate_limiter import Duration, Limiter, Rate
from redis import Redis

from fapi_limiter import DynamicBucketFactory, RateLimiterMiddleware, setup_redis_bucket

app = FastAPI()

limiter = Limiter(
    DynamicBucketFactory(
        setup_redis_bucket([Rate(10, Duration.MINUTE)], redis=Redis(host="localhost"))
    )
)
app.add_middleware(RateLimiterMiddleware, limiter=limiter)

Response Headers

On every allowed request, the following headers are set. With several rates, they describe the rate closest to its limit:

Header	Description
X-RateLimit-Limit	Maximum requests allowed in the window
X-RateLimit-Remaining	Requests remaining in the current window
X-RateLimit-Reset	Unix timestamp (ms) when the window resets

When the limit is exceeded, 429 Too Many Requests is returned and one additional header is set:

Header	Description
Retry-After	Seconds to wait before the client may retry

RateLimiterMiddleware sets these on every response. A RateLimiter dependency can only set them on the response FastAPI creates, so they are missing when a route returns a Response itself or raises an HTTPException. To send them on those too, add this middleware:

from fapi_limiter import RateLimitHeadersMiddleware

app.add_middleware(RateLimitHeadersMiddleware)

Configuration

Both RateLimiterMiddleware and RateLimiter accept the same options:

Parameter	Type	Default	Description
limiter	Limiter	required	A pyrate-limiter Limiter instance
callback	CallbackFunction	default_callback	Called when the limit is exceeded
identifier	Callable[[Request], str]	get_default_identifier	Extracts a key from the request (e.g. a user ID)
blocking	bool	False	Whether to block and wait instead of rejecting
fail_open	bool	False	Allow requests and log when the backend fails

RateLimiterMiddleware also accepts:

Parameter	Type	Default	Description
skip	SkipFunction | None	None	If it returns True, the request bypasses rate limiting

Custom Callback

The callback is invoked when a request is rate-limited. It receives the Request and a headers dict, and should raise an HTTPException. If it returns without raising, the default 429 is sent:

from fastapi import HTTPException, Request
from fapi_limiter import RateLimiterMiddleware


def my_callback(request: Request, headers=None):
    raise HTTPException(status_code=429, detail="Slow down!", headers=headers)


app.add_middleware(RateLimiterMiddleware, limiter=limiter, callback=my_callback)

Custom Identifier

By default RateLimiterMiddleware keys by client IP, because it runs before a route is matched. RateLimiter keys by {request_method}:{client_ip}:{route}, where route is the template (/users/{id}), so changing the path can't get a client a fresh counter. X-Forwarded-For is ignored because clients can forge it. Behind a proxy, have the server resolve the real client address, or every client will share the proxy's limit:

uvicorn main:app --forwarded-allow-ips="<proxy ip>"

You can override the identifier to key by user, API token, etc.:

def by_user(request: Request) -> str:
    return request.headers.get("X-User-Id", "anonymous") + ":" + request.url.path


app.add_middleware(RateLimiterMiddleware, limiter=limiter, identifier=by_user)

Skip Function

Use skip on RateLimiterMiddleware to bypass rate limiting for certain requests (e.g. health checks):

def skip_health(request: Request) -> bool:
    return request.url.path == "/health"


app.add_middleware(RateLimiterMiddleware, limiter=limiter, skip=skip_health)

Credits

• fastapi-limiter by long2ice
• requests-limiter by JWCook

License

MIT