ffsshhttiikk What I do
- Implement rate limiting
- Choose appropriate algorithms
- Handle rate limit headers
- Design tiered limits
- Handle rate limit exceeded
- Distribute rate limits
- Monitor rate limit usage
- Implement custom providers
When to use me
When implementing rate limiting or throttling.
Rate Limiting Algorithms
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional
import asyncio
# Fixed Window
class FixedWindowRateLimiter:
"""
Simple fixed window rate limiting.
Pros: Simple, memory efficient
Cons: Can allow bursts at window boundaries
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.windows: Dict[str, list] = {}
def is_allowed(self, key: str) -> tuple[bool, dict]:
now = datetime.utcnow()
window_start = self._get_window_start(now)
if key not in self.windows:
self.windows[key] = []
# Remove old requests
self.windows[key] = [
t for t in self.windows[key]
if t >= window_start
]
if len(self.windows[key]) >= self.max_requests:
return False, {
"limit": self.max_requests,
"remaining": 0,
"reset": int((window_start + timedelta(seconds=self.window_seconds)).timestamp())
}
self.windows[key].append(now)
return True, {
"limit": self.max_requests,
"remaining": self.max_requests - len(self.windows[key]),
"reset": int((window_start + timedelta(seconds=self.window_seconds)).timestamp())
}
def _get_window_start(self, now: datetime) -> datetime:
seconds_past_window = now.timestamp() % self.window_seconds
return now - timedelta(seconds=seconds_past_window)
# Sliding Window
class SlidingWindowRateLimiter:
"""
Sliding window rate limiting.
Pros: More accurate than fixed window
Cons: More complex, more memory
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: Dict[str, list] = {}
def is_allowed(self, key: str) -> tuple[bool, dict]:
now = datetime.utcnow()
window_start = now - timedelta(seconds=self.window_seconds)
if key not in self.requests:
self.requests[key] = []
# Remove old requests
self.requests[key] = [
t for t in self.requests[key]
if t >= window_start
]
if len(self.requests[key]) >= self.max_requests:
return False, {
"limit": self.max_requests,
"remaining": 0,
"reset": int((now + timedelta(seconds=self.window_seconds)).timestamp())
}
self.requests[key].append(now)
return True, {
"limit": self.max_requests,
"remaining": self.max_requests - len(self.requests[key]),
"reset": int((now + timedelta(seconds=self.window_seconds)).timestamp())
}
# Token Bucket
class TokenBucketRateLimiter:
"""
Token bucket rate limiting.
Pros: Allows bursts, smooth rate limiting
Cons: More complex
"""
def __init__(self, max_tokens: int, refill_rate: float):
self.max_tokens = max_tokens
self.refill_rate = refill_rate # tokens per second
self.tokens: Dict[str, float] = {}
self.last_refill: Dict[str, datetime] = {}
def _refill(self, key: str) -> float:
now = datetime.utcnow()
if key not in self.tokens:
self.tokens[key] = self.max_tokens
self.last_refill[key] = now
return self.tokens[key]
elapsed = (now - self.last_refill[key]).total_seconds()
tokens_to_add = elapsed * self.refill_rate
self.tokens[key] = min(
self.max_tokens,
self.tokens[key] + tokens_to_add
)
self.last_refill[key] = now
return self.tokens[key]
def is_allowed(self, key: str, tokens: int = 1) -> tuple[bool, dict]:
self._refill(key)
if self.tokens[key] >= tokens:
self.tokens[key] -= tokens
return True, {
"limit": self.max_tokens,
"remaining": int(self.tokens[key]),
}
return False, {
"limit": self.max_tokens": 0,
,
"remaining }
API Rate Limiting Middleware
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
class RateLimitMiddleware(BaseHTTPMiddleware):
"""Rate limiting middleware for FastAPI."""
def __init__(
self,
app,
rate_limiter,
exempt_paths: list = None
):
super().__init__(app)
self.rate_limiter = rate_limiter
self.exempt_paths = exempt_paths or []
async def dispatch(self, request: Request, call_next):
# Skip rate limiting for exempt paths
if request.url.path in self.exempt_paths:
return await call_next(request)
# Get client identifier
key = self._get_client_key(request)
# Check rate limit
allowed, headers = self.rate_limiter.is_allowed(key)
if not allowed:
return JSONResponse(
status_code=429,
headers={
**headers,
"Retry-After": str(headers.get("reset", 60)),
},
content={
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Too many requests. Please retry later."
}
}
)
response = await call_next(request)
# Add rate limit headers
for key, value in headers.items():
response.headers[key] = str(value)
return response
def _get_client_key(self, request: Request) -> str:
"""Get unique key for client."""
# Check for API key
api_key = request.headers.get("X-API-Key")
if api_key:
return f"apikey:{api_key}"
# Use user ID if authenticated
if hasattr(request.state, "user_id"):
return f"user:{request.state.user_id}"
# Fall back to IP
client_ip = request.client.host
return f"ip:{client_ip}"
Distributed Rate Limiting with Redis
import redis
from typing import Dict, Any
class RedisRateLimiter:
"""Distributed rate limiter using Redis."""
def __init__(
self,
redis_url: str,
max_requests: int = 100,
window_seconds: int = 60
):
self.redis = redis.from_url(redis_url)
self.max_requests = max_requests
self.window_seconds = window_seconds
def is_allowed(self, key: str) -> tuple[bool, dict]:
now = datetime.utcnow()
window_key = f"ratelimit:{key}:{int(now.timestamp() // self.window_seconds)}"
pipe = self.redis.pipeline()
# Increment counter
pipe.incr(window_key)
# Set expiry
pipe.expire(window_key, self.window_seconds * 2)
# Get current count
results = pipe.execute()
current_count = results[0]
headers = {
"X-RateLimit-Limit": str(self.max_requests),
"X-RateLimit-Remaining": str(max(0, self.max_requests - current_count)),
"X-RateLimit-Reset": str(
int(now.timestamp()) + self.window_seconds
),
}
if current_count > self.max_requests:
return False, headers
return True, headers
def sliding_window(self, key: str) -> tuple[bool, dict]:
"""More accurate sliding window using sorted sets."""
now = datetime.utcnow()
window_start = now - timedelta(seconds=self.window_seconds)
pipe = self.redis.pipeline()
# Remove old entries
pipe.zremrangebyscore(
f"ratelimit:sliding:{key}",
0,
window_start.timestamp() - 1
)
# Add current request
pipe.zadd(
f"ratelimit:sliding:{key}",
{f"{now.timestamp()}:{uuid.uuid4()}": now.timestamp()}
)
# Count requests in window
pipe.zcard(f"ratelimit:sliding:{key}")
# Set expiry
pipe.expire(
f"ratelimit:sliding:{key}",
self.window_seconds * 2
)
results = pipe.execute()
current_count = results[2]
allowed = current_count <= self.max_requests
return allowed, {
"X-RateLimit-Limit": str(self.max_requests),
"X-RateLimit-Remaining": str(
max(0, self.max_requests - current_count)
),
"X-RateLimit-Reset": str(
int(now.timestamp()) + self.window_seconds
),
}
Best Practices
Rate Limiting Best Practices:
1. Choose appropriate algorithm
- Fixed window: simple, forgiving
- Sliding window: more accurate
- Token bucket: burst handling
2. Use tiered limits
- Different limits for different users
- Public vs authenticated
3. Return proper headers
- X-RateLimit-Limit
- X-RateLimit-Remaining
- X-RateLimit-Reset
- Retry-After
4. Handle exceeded gracefully
- Clear error messages
- Appropriate retry headers
5. Monitor and alert
- Track rate limit hits
- Alert on unusual patterns
6. Distribute across instances
- Use shared storage (Redis)
- Consistent hashing
7. Consider endpoint-specific
- Different limits per endpoint
- Expensive operations get lower limits
8. Allow some burst
- Users expect occasional bursts
- Token bucket handles this well
9. Test rate limiting
- Verify limits are enforced
- Test header values
10. Document limits
- API documentation
- Developer portal