-- KEYS[1] = bucket key
-- ARGV[1] = capacity, ARGV[2] = refill_per_sec, ARGV[3] = now (float seconds)
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local state = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(state[1])
local ts = tonumber(state[2])
if tokens == nil then
tokens = capacity
ts = now
end
local elapsed = math.max(0, now - ts)
tokens = math.min(capacity, tokens + elapsed * rate)
local allowed = 0
if tokens >= 1 then
tokens = tokens - 1
allowed = 1
end
redis.call('HSET', KEYS[1], 'tokens', tokens, 'ts', now)
-- expire idle buckets once they would be fully refilled
redis.call('EXPIRE', KEYS[1], math.ceil(capacity / rate) + 1)
return { allowed, tokens }
import time
from dataclasses import dataclass
from pathlib import Path
from redis.asyncio import Redis
_SCRIPT = (Path(__file__).parent / "token_bucket.lua").read_text()
@dataclass
class Decision:
allowed: bool
remaining: int
retry_after: float
class TokenBucketLimiter:
def __init__(self, redis: Redis, capacity: int, refill_per_sec: float, prefix: str = "rl"):
self.capacity = capacity
self.refill_per_sec = refill_per_sec
self.prefix = prefix
self._script = redis.register_script(_SCRIPT)
async def allow(self, identity: str) -> Decision:
key = f"{self.prefix}:{identity}"
now = time.time()
allowed, tokens = await self._script(
keys=[key],
args=[self.capacity, self.refill_per_sec, now],
)
remaining = int(float(tokens))
retry_after = 0.0 if allowed else (1.0 - float(tokens)) / self.refill_per_sec
return Decision(bool(allowed), remaining, round(retry_after, 3))
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from rate_limiter import TokenBucketLimiter
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, limiter: TokenBucketLimiter):
super().__init__(app)
self.limiter = limiter
def _identity(self, request: Request) -> str:
api_key = request.headers.get("x-api-key")
if api_key:
return f"key:{api_key}"
client = request.client.host if request.client else "unknown"
return f"ip:{client}"
async def dispatch(self, request: Request, call_next):
decision = await self.limiter.allow(self._identity(request))
if not decision.allowed:
return JSONResponse(
{"detail": "Rate limit exceeded"},
status_code=429,
headers={"Retry-After": str(decision.retry_after)},
)
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(decision.remaining)
return response
from fastapi import FastAPI
from redis.asyncio import Redis
from middleware import RateLimitMiddleware
from rate_limiter import TokenBucketLimiter
redis = Redis.from_url("redis://localhost:6379/0", decode_responses=False)
# 20 requests burst capacity, refilling at 5 tokens/second
limiter = TokenBucketLimiter(redis, capacity=20, refill_per_sec=5.0)
app = FastAPI()
app.add_middleware(RateLimitMiddleware, limiter=limiter)
@app.get("/search")
async def search(q: str = ""):
return {"query": q, "results": []}
@app.on_event("shutdown")
async def close_redis():
await redis.aclose()
This snippet implements API rate limiting using the token bucket algorithm, enforced as ASGI middleware in FastAPI and backed by Redis so the limit holds across multiple application instances. A token bucket models a bucket that refills at a steady rate up to a fixed capacity; each request consumes one token, and requests are rejected when the bucket is empty. Compared to fixed-window counters, this smooths traffic and allows short bursts up to the bucket's capacity without the boundary spikes fixed windows suffer from.
The core logic lives in token_bucket.lua, a Redis script that reads the stored token count and last-refill timestamp, computes how many tokens should have accrued since the last call, caps the total at capacity, and decrements one token if any remain. Running this as a Lua script is essential: Redis executes it atomically, so the read-modify-write cycle cannot interleave with concurrent requests from other workers. Doing the same in three separate GET/SET calls would create a race where two requests both see the last token and both succeed. The script also sets a TTL so idle buckets expire and do not leak keys.
In rate_limiter.py, TokenBucketLimiter loads and registers the script with register_script, then allow invokes it with the per-client key and the current time. It returns whether the request is permitted plus the remaining tokens and a retry_after hint derived from the refill rate. Passing time from Python keeps the script deterministic and testable, though production systems must ensure clocks are reasonably synchronized.
middleware.py wires this into the request lifecycle via RateLimitMiddleware, which derives a client identity (an API key header falling back to the peer IP), calls limiter.allow, and short-circuits with a 429 and Retry-After header when denied. On success it forwards the request and attaches X-RateLimit-Remaining so clients can self-throttle.
main.py shows the assembly: a Redis connection, one limiter configured with capacity and refill rate, and the middleware added to the app. Key trade-offs to weigh include choosing the identity dimension carefully to avoid punishing users behind shared NATs, and accepting Redis as a dependency in the hot path.
Related snips
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
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
Share this code
Here's the card — post it anywhere.
