fastapi-limiter

To apply rate limiting to a specific route, use the RateLimiter dependency within the dependencies list of your route decorator. This example allows 2 requests every 5 seconds on the / route.

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

from fastapi_limiter.depends import RateLimiter

app = FastAPI()

@app.get(
    "/",
    dependencies=[Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 5))))],
)
async def index():
    return {"msg": "Hello World"}


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)

To apply rate limiting to all routes in your application without adding dependencies to every route, use RateLimiterMiddleware. It accepts the same parameters as RateLimiter: limiter, identifier, callback, blocking, and skip.

from fastapi import FastAPI
from pyrate_limiter import Duration, Limiter, Rate
from fastapi_limiter.middleware import RateLimiterMiddleware

app = FastAPI()

app.add_middleware(
    RateLimiterMiddleware,
    limiter=Limiter(Rate(2, Duration.SECOND * 5)),
)

@app.get("/")
async def index():
    return {"msg": "Hello World"}

Install the package via pip to use rate limiting in your FastAPI applications.

pip install fastapi-limiter

You can stack multiple RateLimiter dependencies on a single route to enforce different rate limits (e.g., a strict limit for short bursts and a broader limit for long durations).

Note: Always place the stricter limiter (the one with the lower seconds/times ratio) first in the dependencies list.

@app.get(
    "/multiple",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))),
        Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 15)))),
    ],
)
async def multiple():
    return {"msg": "Hello World"}

Use the skip parameter in RateLimiter to pass an async callable that determines if a specific request should bypass rate limiting. The callable receives the Request object and must return True to skip.

from fastapi import Request
from fastapi.depends import Depends
from fastapi_limiter.depends import RateLimiter
from pyrate_limiter import Duration, Limiter, Rate

async def skip_logic(request: Request):
    return request.scope["path"] == "/skip"

@app.get(
    "/skip",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)), skip=skip_logic))
    ],
)
async def skip_route():
    return {"This route skips rate limiting"}

You can override the default behavior (which raises a 429 error) by providing a custom callback function. This function is called when the rate limit is exceeded.

from fastapi import HTTPException, status

def custom_callback(*args, **kwargs):
    raise HTTPException(
        status.HTTP_429_TOO_MANY_REQUESTS,
        "Too Many Requests",
    )

By default, RateLimiter identifies requests using ip + path. You can provide a custom identifier callable to limit based on other criteria, such as a userid extracted from headers or tokens.

from typing import Union
from fastapi import Request
from fastapi import WebSocket

async def custom_identifier(request: Union[Request, WebSocket]):
    # Example: Extracting IP from X-Forwarded-For or client host
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        ip = forwarded.split(",")[0]
    elif request.client:
        ip = request.client.host
    else:
        ip = "127.0.0.1"
    return ip + ":" + request.scope["path"]

The RateLimiter dependency accepts the following configuration parameters:

• limiter: A pyrate_limiter.Limiter instance defining the rate limiting rules.
• identifier: A callable to identify the request source (e.g., by user ID or IP). Defaults to ip + path.
• callback: A callable invoked when the limit is exceeded. Defaults to raising an HTTPException with a 429 status code.
• blocking: A boolean indicating whether to block the request when the limit is exceeded. Defaults to False.
• skip: An async callable that takes a request and returns True to bypass rate limiting. Defaults to None.

The RateLimiterMiddleware accepts a skip parameter. This should be an async function that takes a starlette.requests.Request and returns a boolean. If the function returns True, the middleware will call call_next(request) immediately, bypassing all rate limiting checks for that specific request.

async def skip_admin_routes(request):
    # Skip rate limiting if the path starts with /admin
    return request.url.path.startswith("/admin")

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