import time
import uuid
from dataclasses import dataclass
import redis
_LUA = """
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local member = ARGV[4]
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)
if count < limit then
redis.call('ZADD', key, now, member)
redis.call('EXPIRE', key, window)
return {1, limit - count - 1}
end
redis.call('EXPIRE', key, window)
return {0, 0}
"""
@dataclass
class RateLimitResult:
allowed: bool
remaining: int
retry_after: int
class SlidingWindowLimiter:
def __init__(self, client, limit, window):
self.client = client
self.limit = limit
self.window = window
self._script = client.register_script(_LUA)
def is_allowed(self, key):
now = time.time()
member = "{0}-{1}".format(now, uuid.uuid4().hex)
allowed, remaining = self._script(
keys=[key],
args=[now, self.window, self.limit, member],
)
retry_after = 0 if allowed else self.window
return RateLimitResult(
allowed=bool(allowed),
remaining=int(remaining),
retry_after=int(retry_after),
)
from functools import wraps
from flask import request, abort, g, after_this_request
def default_key_func():
user = getattr(g, "current_user", None)
if user is not None:
return "user:{0}".format(user.id)
return "ip:{0}".format(request.remote_addr)
def rate_limit(limiter, scope, key_func=default_key_func):
def wrapper(view):
@wraps(view)
def guarded(*args, **kwargs):
key = "ratelimit:{0}:{1}".format(scope, key_func())
result = limiter.is_allowed(key)
@after_this_request
def add_headers(response):
response.headers["X-RateLimit-Limit"] = str(limiter.limit)
response.headers["X-RateLimit-Remaining"] = str(result.remaining)
if not result.allowed:
response.headers["Retry-After"] = str(result.retry_after)
return response
if not result.allowed:
abort(429, description="Rate limit exceeded for {0}".format(scope))
return view(*args, **kwargs)
return guarded
return wrapper
import redis
from flask import Flask, jsonify
from rate_limiter import SlidingWindowLimiter
from decorator import rate_limit
app = Flask(__name__)
_redis = redis.Redis(host="localhost", port=6379, db=0)
search_limiter = SlidingWindowLimiter(_redis, limit=30, window=60)
export_limiter = SlidingWindowLimiter(_redis, limit=3, window=3600)
@app.route("/api/search")
@rate_limit(search_limiter, scope="search")
def search():
return jsonify(results=[])
@app.route("/api/export", methods=["POST"])
@rate_limit(export_limiter, scope="export")
def export():
return jsonify(status="queued"), 202
@app.errorhandler(429)
def too_many_requests(err):
return jsonify(error="too_many_requests", detail=err.description), 429
if __name__ == "__main__":
app.run()
This snippet implements a per-user rate limiter using a sliding-window counter backed by Redis. A sliding window fixes the classic problem of fixed-window counters, where a burst at the end of one window plus a burst at the start of the next can let through nearly double the intended limit. Instead, the algorithm keeps a sorted set of request timestamps per key and counts only those falling inside the trailing window seconds, so the boundary moves continuously with time.
The SlidingWindowLimiter class in rate_limiter.py wraps the whole operation in a single Redis Lua script. Doing the read-modify-write inside Lua matters: ZREMRANGEBYSCORE, ZCARD, and the conditional ZADD all run atomically on the server, so two concurrent requests can never both read a stale count and both slip past the limit. The script trims entries older than now - window, counts what remains, and only records the new request when the count is under limit; it returns both the allow/deny decision and the number of remaining slots. EXPIRE keeps idle keys from leaking memory.
The is_allowed method loads the script once via register_script and returns a small RateLimitResult dataclass carrying allowed, remaining, and a computed retry_after. Time is passed in from Python rather than read inside Lua so the logic stays testable and deterministic.
In decorator.py, rate_limit adapts the limiter to Flask. It resolves a per-user identity through a pluggable key_func (defaulting to the authenticated user id, falling back to the client IP), builds a namespaced Redis key, and aborts with 429 when the window is full. It always sets X-RateLimit-Limit, X-RateLimit-Remaining, and Retry-After headers so clients can back off intelligently, using Flask's after_this_request to attach them to the eventual response.
app.py wires it together, sharing one limiter instance and applying different budgets per route. The trade-offs worth noting: the sorted set stores one member per request, so very high-traffic keys cost more memory than a bare counter, and the limiter depends on Redis availability — a fail-open or fail-closed policy on connection errors is a deliberate decision the caller should make.
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
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
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
