API Rate Limiter & Retry Engine
Build the traffic control every public API runs: a token bucket, a per-user sliding window, exponential backoff for flaky calls, and middleware that answers 429 with an honest Retry-After.
How it works in the real world
Every serious API (Stripe, GitHub, the Claude API) protects itself from overload, and every serious client copes with being told to slow down:
- Token bucket: a bucket refills at a steady rate and each request spends a token, allowing short bursts but a fixed long-run rate.
- Sliding window: count each user's requests in the last N seconds for fair per-user limits.
- Backoff: when a call fails, wait 1s, 2s, 4s... before retrying, capped, so a struggling server gets room to recover.
- Middleware: put it in front of every request and reply
429 Too Many Requestswith aRetry-Afterheader.
Time is passed in explicitly rather than read from a clock, which is also how you would make this code testable in production.
Token bucket
Finish TokenBucket: it starts full (capacity tokens), refills rate tokens per second up to capacity, and allow(now) spends one token if available. Print t=TIME ALLOW or t=TIME DENY for each request time.
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate); self.last = now; then if self.tokens >= 1: spend one and return True.
class TokenBucket:
def __init__(self, capacity, rate):
self.capacity = capacity
self.rate = rate
self.tokens = capacity
self.last = 0.0
def allow(self, now):
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
self.last = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
bucket = TokenBucket(capacity=3, rate=1)
for t in [0, 0.1, 0.2, 0.3, 1.0, 1.5, 2.6, 2.7, 6.0, 6.1, 6.2, 6.3]:
print(f"t={t} {'ALLOW' if bucket.allow(t) else 'DENY'}")
Run your code to check this step…
Per-user sliding window
Limit each user to LIMIT requests in any WINDOW-second span. Keep a deque of timestamps per user, drop the ones older than the window, and print USER@TIME ok or USER@TIME limited. One user's traffic must never block another.
q = seen[user]; while q and q[0] <= now - WINDOW: q.popleft(). If len(q) < LIMIT, append now and return True.
from collections import defaultdict, deque
LIMIT, WINDOW = 2, 10
seen = defaultdict(deque)
def allow(user, now):
q = seen[user]
while q and q[0] <= now - WINDOW:
q.popleft()
if len(q) < LIMIT:
q.append(now)
return True
return False
requests = [("ana", 0), ("ana", 3), ("ana", 5), ("ben", 5), ("ben", 6), ("ana", 10), ("ana", 13), ("ben", 14)]
for user, now in requests:
print(f"{user}@{now} {'ok' if allow(user, now) else 'limited'}")
Run your code to check this step…
Retry with exponential backoff
Write call_with_retry(fn, retries=5, base=1, cap=8): on ConnectionError, wait min(cap, base * 2 ** attempt) seconds (recorded, not slept) and try again; any other exception is not retried. Print every attempt, then the result and total wait. Run it against a call that fails three times and one that always fails.
for attempt in range(retries + 1): try: return fn(), waited. except ConnectionError as exc: if attempt == retries: print("giving up") and return; else compute the delay, add it to waited and print it.
def flaky(fail_times):
state = {"calls": 0}
def call():
state["calls"] += 1
if state["calls"] <= fail_times:
raise ConnectionError(f"timeout #{state['calls']}")
return "200 OK"
return call
def call_with_retry(fn, retries=5, base=1, cap=8):
waited = 0
for attempt in range(retries + 1):
try:
return fn(), waited
except ConnectionError as exc:
if attempt == retries:
print(f" {exc}: giving up after {attempt + 1} attempts")
return None, waited
delay = min(cap, base * 2 ** attempt)
waited += delay
print(f" {exc}: retrying in {delay}s")
for fails in [3, 9]:
result, waited = call_with_retry(flaky(fails))
print(f"result: {result}, waited {waited}s")
Run your code to check this step…
Rate-limit middleware
Put a token bucket per API key in front of handler. Unknown keys get 401. A key out of tokens gets 429 with Retry-After set to the whole seconds until one token is back (math.ceil). Print one line per request.
bucket = buckets.setdefault(key, TokenBucket(*PLANS[key])); bucket.refill(now). If bucket.tokens < 1, the wait is (1 - bucket.tokens) / bucket.rate seconds.
import math
class TokenBucket:
def __init__(self, capacity, rate):
self.capacity, self.rate = capacity, rate
self.tokens, self.last = capacity, 0.0
def refill(self, now):
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
self.last = now
PLANS = {"key-free": (2, 0.5), "key-pro": (5, 5)}
buckets = {}
def handler(path):
return 200, {}, f"data for {path}"
def middleware(key, path, now):
if key not in PLANS:
return 401, {}, "unknown API key"
bucket = buckets.setdefault(key, TokenBucket(*PLANS[key]))
bucket.refill(now)
if bucket.tokens < 1:
wait = math.ceil((1 - bucket.tokens) / bucket.rate)
return 429, {"Retry-After": wait}, "slow down"
bucket.tokens -= 1
return handler(path)
traffic = [("key-free", "/a", 0), ("key-free", "/b", 0.5), ("key-free", "/c", 1.0),
("key-free", "/d", 3.0), ("key-pro", "/a", 1.0), ("nope", "/a", 1.0)]
for key, path, now in traffic:
status, headers, body = middleware(key, path, now)
extra = f" Retry-After: {headers['Retry-After']}" if "Retry-After" in headers else ""
print(f"{now:>4} {key:<9} {path} -> {status}{extra} {body}")
Run your code to check this step…
Project shipped!
You just built a complete, working api rate limiter & retry engine: the same architecture running in production software. Pick your next build →
Take it further
Run it in front of a real web app, then make it work across several servers. Nothing below is graded: it runs on your own machine, not in this tab.
Stretch goals
- Wrap a Flask or FastAPI app with this middleware and hammer it with a loop of
requests.getcalls. - Store bucket state in Redis so every server instance shares the same limits.
- Add random jitter to the backoff (
random.uniform(0, delay)) and explain why it stops many clients retrying in lockstep. - Read the
Retry-Afterheader in your client and wait exactly that long instead of guessing.
middleware_flask.py: the step 4 limiter in a real Flask app
# pip install flask
import math
import time
from flask import Flask, jsonify, request
app = Flask(__name__)
PLANS = {"key-free": (2, 0.5), "key-pro": (5, 5)}
buckets = {} # key -> [tokens, last_time]
@app.before_request
def rate_limit():
key = request.headers.get("X-API-Key", "")
if key not in PLANS:
return jsonify(error="unknown API key"), 401
capacity, rate = PLANS[key]
tokens, last = buckets.get(key, [capacity, time.monotonic()])
now = time.monotonic()
tokens = min(capacity, tokens + (now - last) * rate)
if tokens < 1:
buckets[key] = [tokens, now]
wait = math.ceil((1 - tokens) / rate)
return jsonify(error="slow down"), 429, {"Retry-After": str(wait)}
buckets[key] = [tokens - 1, now]
@app.get("/data")
def data():
return jsonify(ok=True)
if __name__ == "__main__":
app.run(port=8000)
Ask an AI to review your build
I implemented a token bucket, a per-user sliding window limiter, exponential backoff and a Flask middleware that returns 429 with Retry-After. Here is my code: [paste it]. Where would it break under real load (multiple processes, clock changes, memory growth per key)? Explain the trade-offs between token bucket and sliding window for a public API.