python 38 lines · 7 steps

A token-bucket rate limiter in Flask

A before_request hook throttles clients by refilling and draining a per-key token bucket on every incoming request.

Explained by highlit
Flask Intermediate rate-limiting token-bucket middleware thread-safety concurrency
1import time
2import threading
3from flask import Flask, request, jsonify, g
4 
5app = Flask(__name__)
6 
7RATE = 5.0
8CAPACITY = 10
9_buckets = {}
10_lock = threading.Lock()
11 
12 
13def _consume(key, tokens=1):
14 now = time.monotonic()
15 with _lock:
16 tokens_available, last = _buckets.get(key, (CAPACITY, now))
17 tokens_available = min(CAPACITY, tokens_available + (now - last) * RATE)
18 if tokens_available < tokens:
19 deficit = tokens - tokens_available
20 _buckets[key] = (tokens_available, now)
21 return False, deficit / RATE
22 _buckets[key] = (tokens_available - tokens, now)
23 return True, 0.0
24 
25 
26@app.before_request
27def throttle():
28 key = request.headers.get("X-API-Key") or request.remote_addr
29 allowed, retry_after = _consume(key)
30 if not allowed:
31 response = jsonify(
32 error="rate_limit_exceeded",
33 message="Too many requests, slow down.",
34 )
35 response.status_code = 429
36 response.headers["Retry-After"] = str(int(retry_after) + 1)
37 return response
38 g.rate_limit_key = key
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A token bucket lets you allow short bursts while capping the sustained request rate to a fixed refill speed.
  2. 2Refilling lazily from elapsed time avoids background timers — you compute tokens only when a request arrives.
  3. 3Shared mutable state touched by concurrent requests needs a lock to keep the read-modify-write atomic.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps

Share this explainer

Here's the card — post it anywhere.

A token-bucket rate limiter in Flask — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code