import time
import threading
from collections import deque, defaultdict
class SlidingWindowLimiter:
def __init__(self, limit, window):
self.limit = limit
self.window = float(window)
self._buckets = defaultdict(deque)
self._lock = threading.Lock()
def hit(self, key):
now = time.monotonic()
cutoff = now - self.window
with self._lock:
events = self._buckets[key]
while events and events[0] <= cutoff:
events.popleft()
if len(events) >= self.limit:
reset = events[0] + self.window
return False, 0, reset
events.append(now)
remaining = self.limit - len(events)
reset = events[0] + self.window
return True, remaining, reset
import time
import math
from functools import wraps
from flask import request, abort, make_response
from rate_limit import SlidingWindowLimiter
def _client_key(key_func):
if key_func is not None:
return key_func()
return request.headers.get("X-Forwarded-For", request.remote_addr) or "anonymous"
def _apply_headers(response, limit, remaining, reset):
response.headers["X-RateLimit-Limit"] = str(limit)
response.headers["X-RateLimit-Remaining"] = str(max(remaining, 0))
response.headers["X-RateLimit-Reset"] = str(int(math.ceil(reset)))
return response
def rate_limit(limit, window, key_func=None):
limiter = SlidingWindowLimiter(limit, window)
def decorator(view):
@wraps(view)
def wrapper(*args, **kwargs):
key = _client_key(key_func)
allowed, remaining, reset = limiter.hit(key)
if not allowed:
retry_after = max(int(math.ceil(reset - time.monotonic())), 1)
response = make_response({"error": "rate limit exceeded"}, 429)
response.headers["Retry-After"] = str(retry_after)
_apply_headers(response, limit, remaining, reset)
abort(response)
response = make_response(view(*args, **kwargs))
return _apply_headers(response, limit, remaining, reset)
return wrapper
return decorator
from flask import Flask, jsonify, request
from decorators import rate_limit
app = Flask(__name__)
def _api_key():
return request.headers.get("X-API-Key") or request.remote_addr
@app.route("/api/search")
@rate_limit(limit=10, window=60, key_func=_api_key)
def search():
return jsonify(results=[], query=request.args.get("q", ""))
@app.route("/api/export")
@rate_limit(limit=2, window=60, key_func=_api_key)
def export():
return jsonify(status="queued", format=request.args.get("format", "csv"))
if __name__ == "__main__":
app.run(debug=True)
This snippet builds a request rate limiter for Flask using a hand-rolled sliding-window algorithm stored in process memory, with no external dependencies like Redis. It is useful for single-process deployments, internal tools, or prototypes where a full distributed limiter would be overkill, and it doubles as a clear illustration of how sliding windows and token accounting actually work.
The rate_limit.py module defines SlidingWindowLimiter, which keeps a deque of request timestamps per client key. On each hit, timestamps older than the window are popped from the left, so the deque only ever holds events inside the current window; the remaining length is the request count. This gives a true sliding window rather than the cheaper fixed-window approach, which suffers from burst doubling at the boundary between two windows. A threading.Lock guards each mutation because Flask's default WSGI servers can serve requests on multiple threads, and an unprotected deque would corrupt under concurrent hit calls. The hit method returns both an allow/deny decision and metadata (remaining and reset) so callers can populate standard headers.
The decorators.py file wraps the limiter in a rate_limit decorator factory. It uses functools.wraps to preserve the view's identity, derives a client key from an override function or the remote IP, and on rejection raises a 429 via abort with Retry-After. The interesting part is _apply_headers: it attaches X-RateLimit-* headers to every response, including allowed ones, so clients can self-throttle before hitting the wall. A per-endpoint SlidingWindowLimiter is created once at decoration time, keeping state isolated between routes.
In app.py, the decorator is applied to two routes with different budgets, showing how limits are tuned per endpoint. A shared after_request hook is unnecessary because the decorator manages its own headers. The main trade-off is honesty: because state lives in memory, limits reset on restart and are not shared across workers, so a multi-process gunicorn deployment would let each worker grant the full quota. For that case a shared store is required, but the interface here is designed so swapping the backend is straightforward. The pitfall to watch is unbounded growth of the per-key dictionary; a production version would evict idle keys on a timer.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
type User {
id: ID!
name: String!
email: String!
posts: [Post!]!
createdAt: String!GraphQL API with Spring Boot
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
Share this code
Here's the card — post it anywhere.
