Rate Limiting APIs Without Flask-Limiter
Rate limiting protects your API from abuse, bot traffic, and accidental DOS attacks. You don't need Flask-Limiter — here's a clean DIY implementation.
In-Memory Sliding Window
import time
import threading
from collections import defaultdict, deque
_lock = threading.Lock()
_hits = defaultdict(deque) # key → deque of timestamps
def is_rate_limited(key: str, limit: int = 60, window: int = 60) -> bool:
"""Return True if `key` has exceeded `limit` requests in `window` seconds."""
now = time.time()
cutoff = now - window
with _lock:
dq = _hits[key]
# Remove old timestamps outside the window
while dq and dq[0] < cutoff:
dq.popleft()
if len(dq) >= limit:
return True
dq.append(now)
return False
Using It in Flask
from flask import request, jsonify
def rate_limit(limit=60, window=60, key_func=None):
"""Decorator: rate-limit by IP (or custom key)."""
import functools
def decorator(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
key = (key_func() if key_func else request.remote_addr) or 'global'
if is_rate_limited(f'rl:{key}:{f.__name__}', limit, window):
return jsonify({'error': 'Too many requests'}), 429
return f(*args, **kwargs)
return wrapper
return decorator
@app.route('/api/search')
@rate_limit(limit=30, window=60) # 30 requests per minute per IP
def search():
...
Redis-Backed (Distributed)
For multi-process / multi-server deployments, share state in Redis:
import redis
import time
r = redis.Redis.from_url(os.environ.get('REDIS_URL', 'redis://localhost:6379/0'))
def is_rate_limited_redis(key: str, limit: int, window: int) -> bool:
pipe = r.pipeline()
now = int(time.time() * 1000) # milliseconds
pipe.zremrangebyscore(key, '-inf', now - window * 1000)
pipe.zadd(key, {str(now): now})
pipe.zcard(key)
pipe.expire(key, window)
_, _, count, _ = pipe.execute()
return count > limit
Returning Rate Limit Headers
Good APIs tell clients how many requests they have left:
@app.after_request
def add_rate_headers(response):
# Add these headers so clients can back off gracefully
response.headers['X-RateLimit-Limit'] = '60'
response.headers['X-RateLimit-Remaining'] = '...' # calculate
response.headers['X-RateLimit-Reset'] = str(int(time.time()) + 60)
return response
Per-User vs Per-IP
def get_rate_key():
if 'user_id' in session:
return f"user:{session['user_id']}"
return f"ip:{request.remote_addr}"
In-memory works for single-server apps. Redis works everywhere.
0 Comments
Join the conversation
No comments yet. Be the first!