Input
{"task":"Add rate limiting to my FastAPI endpoint so users can't spam it."}
Output
**What are you protecting, and what counts as "spam"?**
Before I write anything, I need two numbers from you:
- **Limit** – e.g. 10 requests…
- **Window** – …per minute? per hour?
And one architecture question:
- **Single process** (one Uvicorn worker, local dev/small app) → in-memory is fine.
- **Multiple workers / deployed behind a load balancer** → needs Redis, otherwise each worker has its own counter and the limit is multiplied by worker count.
---
## Meanwhile – the smallest complete answer for each case
### Case A · Single process · `slowapi` (most common FastAPI pattern)
```bash
pip install slowapi
```
```python
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import FastAPI, Request
limiter = Limiter(key_func=get_remote_address) # rate-limit by IP
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/your-endpoint")
@limiter.limit("10/minute") # ← change this
async def your_endpoint(request: Request): # request param required by slowapi
...
```
Returns **429 Too Many Requests** automatically. In-memory, zero infra.
---
### Case B · Multiple workers → add Redis
```bash
pip install slowapi redis
```
```python
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import FastAPI, Request
limiter = Limiter(
key_func=get_remote_address,
storage_uri="redis://localhost:6379", # ← your Redis URL
)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/your-endpoint")
@limiter.limit("10/minute")
async def your_endpoint(request: Request):
...
```
Everything else is identical. Swap in the URI and you're done.
---
### If you're behind Nginx / a proxy
`get_remote_address` reads `request.client.host`. Behind a proxy that sets `X-Forwarded-For`, replace the key function:
```python
def get_forwarded_ip(request: Request) -> str:
return request.headers.get("x-forwarded-for", request.client.host).split(",")[0].strip()
limiter = Limiter(key_func=get_forwarded_ip)
```
`shortcut: trusts X-Forwarded-For header blindly; set trusted_hosts in your proxy config or validate the header if spoofing is a concern.`
---
**Skipped:** auth-based limiting (per user/token instead of per IP), dynamic limits by plan/role, custom 429 response body — none asked for.
**Risk:** IP-based limiting blocks all users behind a shared NAT (office, university) if one of them spams.
Why it failed: All assertions passed
History on main
