#243: Rate Limit in FastAPI

A simple way to protect our API is to limit the number of calls a client can make in a given timespan. For FastAPI, we can use the SlowApi package to prevent our application from being flooded with requests.

Installation

We can install the SlowApi package with this command:

pip install slowapi

Limit a URL

With this minimalistic FastAPI application we can limit the access to the /max/{id} URL to 3 calls per minute:

from datetime import datetime
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


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


@app.get("/max/{id}")
@limiter.limit("3/minute")
async def myendpoint(request: Request, id: int):
    return {"id":id, "viewed":datetime.now()}

We can verify that the behaviour works as expected with this test:

from fastapi.testclient import TestClient
from .main_endpoint import app

client = TestClient(app)

def test_limit_url():
    response = client.get("/max/1")
    assert response.status_code == 200

    response = client.get("/max/1")
    assert response.status_code == 200

    response = client.get("/max/1")
    assert response.status_code == 200

    response = client.get("/max/1")
    assert response.status_code == 429
    assert response.json()["error"] == "Rate limit exceeded: 3 per 1 minute"

    response = client.get("/max/2")
    assert response.status_code == 200

When we call the same id on our /max endpoint for the 4th time in a minute, we get this exception:

{ "error": "Rate limit exceeded: 3 per 1 minute" }

Since the default behaviour is per URL, we get a different counter for /max/1 than we get for /max/2. The URL differ and so does our counter for the limit.

Limit an endpoint

If we want to have a limit that works for the whole endpoint and not just the specific URL as above, we can add the parameter key_style="endpoint" in our limiter:

...

limiter = Limiter(key_func=get_remote_address, key_style="endpoint")

...

With this setting the calls to /max/1 and /max/2 count towards the same limit and as soon we hit the endpoint 3 times, we get the exception no matter what parameter we use.

Global limits

If we want to set a global limit for our clients that includes all endpoints, we can use this code:

from datetime import datetime
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
from slowapi.middleware import SlowAPIMiddleware


limiter = Limiter(key_func=get_remote_address, application_limits=["3/minute"])
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)


@app.get("/max/{id}")
async def myendpoint(request: Request, id: int):
    return {"id":id, "viewed":datetime.now()}


@app.get("/hi")
async def hello(request: Request):
    return {"message": "Hello"}

This adds a middleware to FastAPI that handles the limits and we do not need to add the @limit decorator to our endpoints. Be aware that we must use the application_limits parameter to get a global limit. If you use the default_limits parameter, you get a limit per URL even when you use the middleware.

To check if this really covers all endpoints and parameters, we can use this test:

from fastapi.testclient import TestClient
from .main_global import app

client = TestClient(app)

def test_limit_global():
    response = client.get("/max/1")
    assert response.status_code == 200

    response = client.get("/max/2")
    assert response.status_code == 200

    response = client.get("/max/3")
    assert response.status_code == 200

    response = client.get("/max/1")
    assert response.status_code == 429
    assert response.json()["error"] == "Rate limit exceeded: 3 per 1 minute"

    response = client.get("/max/4")
    assert response.status_code == 429
    assert response.json()["error"] == "Rate limit exceeded: 3 per 1 minute"

    response = client.get("/hi")
    assert response.status_code == 429
    assert response.json()["error"] == "Rate limit exceeded: 3 per 1 minute"

If we want to exclude an endpoint from the limits, we can add the @limiter.exempt decorator.

Limitations and options

There are two main limitations in SlowApi we need to be aware of:

We must pass Request as a parameter to our endpoints.
The @limiter decorator must come below the @app decorator, or else the limit will not work.

Especially the second point is easy to overlook and can lead to a long debugging session.

If we do not specify a storage_uri, SlowApi uses an in-memory cache to track the limits. If we want to use Redis instead, we can point our Limiter to our Redis server:

limiter = Limiter(key_func=get_remote_address, storage_uri="redis://<host>:<port>/n")

The documentation for SlowApi explains the necessary details to connect to Redis and how we can use the other options.