Chapter 29, Advanced
Request limitation
Without limits, a faulty (or malicious) client can take down your API, exhaust a paid quota, or test millions of passwords. Rate limiting is the simplest defense against all three.
The idea
Instead of serving everything, the API accepts a maximum number of requests per time window, per client (for example, 5 per minute per IP address). Exceeded the limit, it returns 429 Too Many Requests, and preferably, the header Retry-After, which tells when to try again.
With the `slowapi` library
The slowapi is the most used library for this in FastAPI. The key get_remote_address identifies the client by IP, and the decorator @limiter.limit sets the limit for each route. The route function needs to receive request: Request, which the slowapi uses to know who made the call:
uv add slowapi
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
@app.exception_handler(RateLimitExceeded)
async def limite_excedido(request: Request, exc: RateLimitExceeded):
return JSONResponse(
status_code=429,
content={"detail": "Muitas requisições"},
headers={"Retry-After": "60"},
)
@app.get("/dados")
@limiter.limit("5/minute")
def dados(request: Request):
return {"mensagem": "sucesso"}
cliente = TestClient(app)
respostas = [cliente.get("/dados") for _ in range(7)]
print([r.status_code for r in respostas])
print(respostas[-1].json(), respostas[-1].headers["retry-after"])
[200, 200, 200, 200, 200, 429, 429]
{'detail': 'Muitas requisições'} 60
The first five passed, and from the sixth onwards came 429. The order of decorators matters: @app.get outside, @limiter.limit inside.
The algorithms behind
There are three common ways to count, and they behave differently at the edges:
| Algorithm | How it works | Feature |
|---|---|---|
| Fixed window | Counts per "clock" minute and resets | Simple, but allows a double burst at the window's edge |
| Sliding window | Counts the last 60 seconds from now | Fairer, keeps more state |
| Token bucket (token bucket) | A bucket receives tokens at a fixed rate, and each request consumes one | Allows small bursts and limits the average |
The token bucket is the most used, and fits in a few lines. As in caching, the clock is a parameter, so the test doesn't have to wait:
import time
from typing import Callable
class BaldeDeFichas:
def __init__(self, capacidade: float, fichas_por_segundo: float, relogio: Callable[[], float] = time.monotonic):
self.capacidade = capacidade
self.taxa = fichas_por_segundo
self.relogio = relogio
self.fichas = capacidade
self.ultimo = relogio()
def permitir(self, custo: float = 1.0) -> bool:
agora = self.relogio()
self.fichas = min(self.capacidade, self.fichas + (agora - self.ultimo) * self.taxa)
self.ultimo = agora
if self.fichas >= custo:
self.fichas -= custo
return True
return False
class Relogio:
def __init__(self):
self.agora = 0.0
def __call__(self) -> float:
return self.agora
r = Relogio()
balde = BaldeDeFichas(capacidade=3, fichas_por_segundo=1, relogio=r)
print([balde.permitir() for _ in range(4)])
r.agora = 2.0
print([balde.permitir() for _ in range(3)])
[True, True, True, False]
[True, True, False]
The bucket starts with 3 tokens: three requests pass, the fourth is denied. After 2 seconds, it received 2 tokens (1 per second), and two of the three subsequent requests pass.
A self-imposed limit, like dependency
With the bucket, you can write the limiter as a dependency (chapter 13), which only applies to the routes that request it, and can use any key, not just the IP, but the authenticated user:
from fastapi import Depends, HTTPException
baldes: dict[str, BaldeDeFichas] = {}
def limitar_login(request: Request):
chave = request.client.host if request.client else "desconhecido"
balde = baldes.setdefault(chave, BaldeDeFichas(capacidade=3, fichas_por_segundo=3 / 60))
if not balde.permitir():
raise HTTPException(429, "Muitas tentativas de login", headers={"Retry-After": "20"})
app2 = FastAPI()
@app2.post("/login", dependencies=[Depends(limitar_login)])
def login():
return {"ok": True}
c2 = TestClient(app2)
print([c2.post("/login").status_code for _ in range(5)])
[200, 200, 200, 429, 429]
Limit the login is one of the most important applications: without a limit, an attacker tests thousands of passwords per minute against the route in chapter 20.
Where this count can fail
Behind a proxy, the
request.client.hostis the address of the proxy, and all clients appear as one. Configure the server to trust the proxy headers (--proxy-headersand--forwarded-allow-ips), and only of them: the headerX-Forwarded-Forcan be forged by anyone speaking directly with the server. With multiple processes, each counts its own limit (a limit of 5 per minute with 4 processes becomes up to 20). For a global limit, the counter needs to be in shared storage, like Redis, which theslowapisupports by configuration.
Exercise 1
Three attempts per minute
Create a BaldeDeFichas with a capacity of 3 and replenishing 1 token every 20 seconds, with a fake clock. Check that 3 attempts pass, the 4th is denied, and that after 20 seconds a new one passes.