Input
{"task":"Add rate limiting to my FastAPI endpoint so users can't spam it."}
Output
# Rate Limiting in FastAPI
Here's a robust rate limiting implementation using a sliding window algorithm with multiple storage options.
## 1. Install Dependencies
```bash
pip install fastapi redis slowapi
```
## 2. Core Rate Limiter Implementation
```python
# rate_limiter.py
import time
import asyncio
from collections import defaultdict, deque
from typing import Optional
from fastapi import Request, HTTPException, status
from functools import wraps
class InMemoryRateLimiter:
"""
Sliding window rate limiter.
Tracks request timestamps per identifier (IP, user ID, etc.)
"""
def __init__(self):
self.requests: dict[str, deque] = defaultdict(deque)
self._lock = asyncio.Lock()
async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int,
) -> tuple[bool, dict]:
"""
Returns (is_allowed, rate_limit_info).
Removes expired timestamps and checks current count.
"""
async with self._lock:
now = time.time()
window_start = now - window_seconds
request_times = self.requests[key]
# Remove timestamps outside the window
while request_times and request_times[0] < window_start:
request_times.popleft()
current_count = len(request_times)
remaining = max_requests - current_count
if current_count >= max_requests:
# Calculate when the oldest request expires
retry_after = int(request_times[0] - window_start) + 1
return False, {
"limit": max_requests,
"remaining": 0,
"reset_after": retry_after,
"window_seconds": window_seconds,
}
# Record this request
request_times.append(now)
reset_after = int(request_times[0] - window_start) + window_seconds
return True, {
"limit": max_requests,
"remaining": remaining - 1,
"reset_after": reset_after,
"window_seconds": window_seconds,
}
async def reset(self, key: str):
"""Manually reset limits for a key (useful for testing)."""
async with self._lock:
self.requests.pop(key, None)
# Singleton instance
_limiter = InMemoryRateLimiter()
def get_limiter() -> InMemoryRateLimiter:
return _limiter
```
## 3. Dependency Factory
```python
# dependencies.py
from fastapi import Request, HTTPException, status, Depends
from rate_limiter import get_limiter, InMemoryRateLimiter
def get_client_identifier(request: Request) -> str:
"""
Extract client identifier for rate limiting.
Prefers authenticated user ID, falls back to IP address.
"""
# Use authenticated user ID if available
user = getattr(request.state, "user", None)
if user and hasattr(user, "id"):
return f"user:{user.id}"
# Handle proxied requests (nginx, load balancers)
forwarded_for = request.headers.get("X-Forwarded-For")
if forwarded_for:
client_ip = forwarded_for.split(",")[0].strip()
else:
client_ip = request.client.host if request.client else "unknown"
return f"ip:{client_ip}"
def rate_limit(max_requests: int, window_seconds: int, scope: str = ""):
"""
Dependency factory for rate limiting.
Args:
max_requests: Maximum number of requests allowed
window_seconds: Time window in seconds
scope: Optional scope to namespace limits (e.g. "login", "upload")
Usage:
@app.get("/endpoint", dependencies=[Depends(rate_limit(10, 60))])
"""
async def dependency(
request: Request,
limiter: InMemoryRateLimiter = Depends(get_limiter),
):
identifier = get_client_identifier(request)
endpoint = scope or request.url.path
key = f"{endpoint}:{identifier}"
allowed, info = await limiter.is_allowed(key, max_requests, window_seconds)
#…
Why it failed: All assertions passed
History on main
