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.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
Installation
We can install the SlowApi package with this command:
|
1 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
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:
|
1 2 3 4 5 |
... 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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
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:
|
1 |
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.
Next
With SlowApi we get a quick solution to limit the calls of a client. We can choose to limit a specific endpoint, a URL or cover the whole API. Make sure that you adapt the limit to match the usage pattern of your API – otherwise you will lock out too many clients.
Next week we add a login solution to our application to prevent unauthorised people to mess with our tasks.
2 thoughts on “Python Friday #243: Rate Limit in FastAPI”