600 Reference 68991c98

API Rate Limiting Reference

Comprehensive technical reference for rate limiting algorithms, implementations, and best practices

Table of Contents

1. Fundamentals
2. Rate Limiting Algorithms
3. Distributed Rate Limiting
4. HTTP Headers and Standards
5. Storage Backends
6. Implementation Patterns
7. Limiting Strategies
8. Error Handling and Responses
9. Performance and Scalability
10. Security Considerations
11. Testing and Monitoring
12. Common Anti-Patterns
13. Language-Specific Implementations
14. References and Standards

Fundamentals

What is Rate Limiting?

Rate limiting is a technique for controlling the rate at which clients can make requests to an API or service. It prevents abuse, ensures fair resource allocation, and protects services from being overwhelmed.

Why Rate Limit?

Primary objectives:

1. Prevent abuse: Stop malicious actors from overwhelming your service
2. Ensure fair usage: Distribute resources equitably among users
3. Control costs: Limit expensive operations (database queries, external API calls)
4. Maintain stability: Prevent cascading failures under high load
5. Business monetization: Differentiate service tiers

Rate Limiting vs Throttling

RATE LIMITING
├─ Hard limits enforced
├─ Requests rejected when exceeded
├─ Clear quotas communicated
└─ 429 Too Many Requests response

THROTTLING
├─ Soft limits enforced
├─ Requests slowed/delayed
├─ Queuing mechanisms
└─ May not return errors

Key Metrics

Requests Per Second (RPS):

RPS = Total Requests / Time Period (seconds)

Example: 100 requests in 10 seconds = 10 RPS

Burst Capacity:

• Maximum requests allowed in short time window
• Important for handling traffic spikes
• Typically higher than sustained rate

Window Duration:

• Time period for counting requests
• Common: 1 second, 1 minute, 1 hour, 1 day
• Trade-off: Shorter = smoother, Longer = simpler

Rate Limiting Algorithms

Token Bucket Algorithm

Description: Tokens added to bucket at fixed rate. Each request consumes tokens. Request rejected if insufficient tokens available.

Characteristics:

• Allows bursts (up to bucket capacity)
• Smooth average rate over time
• Good for credit-based systems

Parameters:

• capacity: Maximum tokens in bucket
• refill_rate: Tokens added per second
• tokens_per_request: Tokens consumed per request (usually 1)

Mathematical Model:

tokens(t) = min(capacity, tokens(t-1) + refill_rate * Δt)

allowed = tokens(t) >= tokens_needed
if allowed:
    tokens(t) = tokens(t) - tokens_needed

Python Implementation:

import time
from typing import Optional

class TokenBucket:
    """Token bucket rate limiter

    Allows bursts while maintaining average rate.
    """

    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: Maximum tokens in bucket
            refill_rate: Tokens added per second
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.time()
        self._lock = threading.Lock()

    def _refill(self) -> None:
        """Refill tokens based on elapsed time"""
        now = time.time()
        elapsed = now - self.last_refill

        # Add tokens for time elapsed
        tokens_to_add = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        """Attempt to consume tokens

        Args:
            tokens: Number of tokens to consume

        Returns:
            True if tokens consumed, False if insufficient
        """
        with self._lock:
            self._refill()

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True

            return False

    def get_tokens(self) -> float:
        """Get current token count"""
        with self._lock:
            self._refill()
            return self.tokens

    def wait_time(self, tokens: int = 1) -> float:
        """Calculate wait time for tokens

        Returns:
            Seconds to wait until tokens available
        """
        with self._lock:
            self._refill()

            if self.tokens >= tokens:
                return 0.0

            deficit = tokens - self.tokens
            return deficit / self.refill_rate

# Usage example
limiter = TokenBucket(capacity=100, refill_rate=10)  # 10 tokens/sec

if limiter.consume():
    # Process request
    response = handle_request()
else:
    # Rate limited
    wait = limiter.wait_time()
    return error_response(f"Rate limited. Try again in {wait:.1f} seconds", 429)

Distributed Implementation (Redis + Lua):

import redis
import time

class DistributedTokenBucket:
    """Distributed token bucket using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 capacity: int, refill_rate: float):
        self.redis = redis_client
        self.key = f"rate_limit:token_bucket:{key}"
        self.capacity = capacity
        self.refill_rate = refill_rate

        # Lua script for atomic token bucket operations
        self.script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local tokens_needed = tonumber(ARGV[3])
        local now = tonumber(ARGV[4])

        -- Get current state
        local state = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = tonumber(state[1])
        local last_refill = tonumber(state[2])

        -- Initialize if needed
        if not tokens then
            tokens = capacity
            last_refill = now
        end

        -- Refill tokens
        local elapsed = now - last_refill
        tokens = math.min(capacity, tokens + (elapsed * refill_rate))

        -- Check if enough tokens
        if tokens >= tokens_needed then
            tokens = tokens - tokens_needed
            redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
            redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) * 2)
            return {1, tokens}  -- allowed, remaining
        else
            return {0, tokens}  -- denied, remaining
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def consume(self, tokens: int = 1) -> tuple[bool, float]:
        """Attempt to consume tokens

        Returns:
            (allowed, remaining_tokens)
        """
        now = time.time()

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,  # number of keys
                self.key,
                self.capacity,
                self.refill_rate,
                tokens,
                now
            )

            allowed = bool(result[0])
            remaining = float(result[1])

            return allowed, remaining

        except redis.exceptions.NoScriptError:
            # Script not cached, reload
            self.script_sha = self.redis.script_load(self.script)
            return self.consume(tokens)

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedTokenBucket(redis_client, "user:123", capacity=100, refill_rate=10)

allowed, remaining = limiter.consume()
if allowed:
    print(f"Request allowed. {remaining:.1f} tokens remaining")
else:
    print(f"Rate limited. {remaining:.1f} tokens available")

Best For:

• APIs with bursty traffic patterns
• Credit-based systems
• Flexible rate limiting with burst tolerance

Pros:

• Simple to understand and implement
• Allows bursts while maintaining average rate
• Low memory footprint
• Fast operations

Cons:

• Allows bursts (may not be desired)
• Slightly more complex than fixed window

Leaky Bucket Algorithm

Description: Requests enter a queue (bucket). Processed at fixed rate (leak). Requests rejected if queue full.

Characteristics:

• Smooths traffic to constant rate
• No bursts allowed
• Queue-based approach

Parameters:

• capacity: Maximum queue size
• leak_rate: Requests processed per second

Mathematical Model:

queue_size(t) = max(0, queue_size(t-1) - leak_rate * Δt)

if queue_size(t) < capacity:
    queue_size(t) = queue_size(t) + 1
    allowed = true
else:
    allowed = false

Python Implementation:

import time
import threading
from collections import deque

class LeakyBucket:
    """Leaky bucket rate limiter

    Processes requests at constant rate.
    """

    def __init__(self, capacity: int, leak_rate: float):
        """
        Args:
            capacity: Maximum queue size
            leak_rate: Requests processed per second
        """
        self.capacity = capacity
        self.leak_rate = leak_rate
        self.queue_size = 0.0
        self.last_leak = time.time()
        self._lock = threading.Lock()

    def _leak(self) -> None:
        """Process (leak) requests based on elapsed time"""
        now = time.time()
        elapsed = now - self.last_leak

        # Leak (process) requests
        leaked = elapsed * self.leak_rate
        self.queue_size = max(0.0, self.queue_size - leaked)
        self.last_leak = now

    def allow_request(self) -> bool:
        """Check if request can be queued

        Returns:
            True if space available in queue
        """
        with self._lock:
            self._leak()

            if self.queue_size < self.capacity:
                self.queue_size += 1.0
                return True

            return False

    def get_queue_size(self) -> float:
        """Get current queue size"""
        with self._lock:
            self._leak()
            return self.queue_size

# Usage
limiter = LeakyBucket(capacity=50, leak_rate=5)  # 5 requests/sec

if limiter.allow_request():
    # Queue request
    response = handle_request()
else:
    # Queue full
    return error_response("Rate limited. Queue full.", 429)

Distributed Implementation (Redis):

import redis
import time

class DistributedLeakyBucket:
    """Distributed leaky bucket using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 capacity: int, leak_rate: float):
        self.redis = redis_client
        self.key = f"rate_limit:leaky_bucket:{key}"
        self.capacity = capacity
        self.leak_rate = leak_rate

        self.script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local leak_rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local state = redis.call('HMGET', key, 'size', 'last_leak')
        local size = tonumber(state[1]) or 0
        local last_leak = tonumber(state[2]) or now

        -- Leak based on time elapsed
        local elapsed = now - last_leak
        local leaked = elapsed * leak_rate
        size = math.max(0, size - leaked)

        -- Check if space available
        if size < capacity then
            size = size + 1
            redis.call('HMSET', key, 'size', size, 'last_leak', now)
            redis.call('EXPIRE', key, math.ceil(capacity / leak_rate) * 2)
            return {1, capacity - size}  -- allowed, remaining
        else
            return {0, 0}  -- denied
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def allow_request(self) -> tuple[bool, float]:
        """Check if request can be queued

        Returns:
            (allowed, remaining_capacity)
        """
        now = time.time()

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,
                self.key,
                self.capacity,
                self.leak_rate,
                now
            )

            return bool(result[0]), float(result[1])

        except redis.exceptions.NoScriptError:
            self.script_sha = self.redis.script_load(self.script)
            return self.allow_request()

Best For:

• Protecting downstream services
• Guaranteed constant load
• Queue-based processing

Pros:

• Smooth, predictable rate
• No bursts
• Protects downstream services

Cons:

• No burst tolerance
• May delay requests
• More complex than token bucket

Fixed Window Algorithm

Description: Time divided into fixed windows. Counter increments per request. Counter resets at window boundary.

Characteristics:

• Simple to implement
• Easy to understand
• Allows bursts at boundaries

Parameters:

• limit: Maximum requests per window
• window_duration: Length of time window (seconds)

Mathematical Model:

window_id = floor(timestamp / window_duration)
count[window_id] = count[window_id] + 1

allowed = count[window_id] <= limit

Problem: Boundary Burst:

Window 1 (11:59:00 - 11:59:59): 100 requests at 11:59:59
Window 2 (12:00:00 - 12:00:59): 100 requests at 12:00:00

Result: 200 requests in 1 second (burst at boundary)

Python Implementation:

import time
import threading
from collections import defaultdict

class FixedWindow:
    """Fixed window rate limiter

    Simple per-period limits with boundary bursts.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests per window
            window_seconds: Duration of window in seconds
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters = defaultdict(int)
        self._lock = threading.Lock()

    def _get_window_id(self) -> int:
        """Get current window identifier"""
        return int(time.time() // self.window_seconds)

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            window_id = self._get_window_id()

            # Clean old windows (optional optimization)
            old_windows = [wid for wid in self.counters if wid < window_id - 1]
            for wid in old_windows:
                del self.counters[wid]

            # Check limit
            if self.counters[window_id] < self.limit:
                self.counters[window_id] += 1
                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests in current window"""
        with self._lock:
            window_id = self._get_window_id()
            count = self.counters[window_id]
            return max(0, self.limit - count)

    def get_reset_time(self) -> int:
        """Get timestamp when window resets"""
        window_id = self._get_window_id()
        return int((window_id + 1) * self.window_seconds)

# Usage
limiter = FixedWindow(limit=100, window_seconds=60)  # 100 req/min

if limiter.allow_request():
    response = handle_request()
else:
    reset_time = limiter.get_reset_time()
    retry_after = reset_time - int(time.time())
    return error_response(f"Rate limited. Try in {retry_after}s", 429)

Distributed Implementation (Redis):

import redis
import time

class DistributedFixedWindow:
    """Distributed fixed window using Redis"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limit: int, window_seconds: int):
        self.redis = redis_client
        self.key_prefix = f"rate_limit:fixed_window:{key}"
        self.limit = limit
        self.window_seconds = window_seconds

    def _get_window_key(self) -> str:
        """Get Redis key for current window"""
        window_id = int(time.time() // self.window_seconds)
        return f"{self.key_prefix}:{window_id}"

    def allow_request(self) -> tuple[bool, int, int]:
        """Check if request is allowed

        Returns:
            (allowed, remaining, reset_time)
        """
        key = self._get_window_key()

        # Increment counter atomically
        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window_seconds * 2)  # TTL safety margin
        results = pipe.execute()

        count = results[0]
        allowed = count <= self.limit
        remaining = max(0, self.limit - count)

        # Calculate reset time
        window_id = int(time.time() // self.window_seconds)
        reset_time = int((window_id + 1) * self.window_seconds)

        return allowed, remaining, reset_time

    def get_remaining(self) -> int:
        """Get remaining requests in current window"""
        key = self._get_window_key()
        count = int(self.redis.get(key) or 0)
        return max(0, self.limit - count)

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedFixedWindow(redis_client, "user:123", limit=100, window_seconds=60)

allowed, remaining, reset_time = limiter.allow_request()
if allowed:
    # Add headers
    response.headers['X-RateLimit-Limit'] = str(limiter.limit)
    response.headers['X-RateLimit-Remaining'] = str(remaining)
    response.headers['X-RateLimit-Reset'] = str(reset_time)
else:
    retry_after = reset_time - int(time.time())
    return error_response(f"Rate limited. Reset in {retry_after}s", 429)

Best For:

• Simple quota enforcement
• Easy user communication
• Low complexity requirements

Pros:

• Extremely simple
• Low memory usage
• Fast operations
• Easy to explain to users

Cons:

• Allows boundary bursts (2x limit in 1 second)
• Not smooth rate limiting

Sliding Window Algorithm

Description: Track request timestamps. Count requests in rolling time window. Remove old requests outside window.

Characteristics:

• Smooth rate limiting
• No boundary bursts
• Higher memory usage

Parameters:

• limit: Maximum requests in window
• window_duration: Rolling window size (seconds)

Mathematical Model:

window_start = current_time - window_duration
requests_in_window = count(requests where timestamp >= window_start)

allowed = requests_in_window < limit

Python Implementation:

import time
import threading
from collections import deque

class SlidingWindow:
    """Sliding window rate limiter

    Smooth rate limiting without boundary bursts.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests in window
            window_seconds: Rolling window duration
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.requests = deque()
        self._lock = threading.Lock()

    def _clean_old_requests(self, now: float) -> None:
        """Remove requests outside window"""
        window_start = now - self.window_seconds

        while self.requests and self.requests[0] < window_start:
            self.requests.popleft()

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)

            if len(self.requests) < self.limit:
                self.requests.append(now)
                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests in window"""
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)
            return max(0, self.limit - len(self.requests))

    def get_oldest_request_time(self) -> float:
        """Get timestamp of oldest request in window"""
        with self._lock:
            now = time.time()
            self._clean_old_requests(now)

            if self.requests:
                return self.requests[0]
            return now

# Usage
limiter = SlidingWindow(limit=100, window_seconds=60)

if limiter.allow_request():
    response = handle_request()
else:
    # Calculate when next request allowed
    oldest = limiter.get_oldest_request_time()
    retry_after = int(oldest + limiter.window_seconds - time.time())
    return error_response(f"Rate limited. Try in {retry_after}s", 429)

Distributed Implementation (Redis Sorted Sets):

import redis
import time
import uuid

class DistributedSlidingWindow:
    """Distributed sliding window using Redis sorted sets"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limit: int, window_seconds: int):
        self.redis = redis_client
        self.key = f"rate_limit:sliding_window:{key}"
        self.limit = limit
        self.window_seconds = window_seconds

        # Lua script for atomic operations
        self.script = """
        local key = KEYS[1]
        local limit = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])
        local request_id = ARGV[4]

        local window_start = now - window

        -- Remove old requests
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)

        -- Count requests in window
        local count = redis.call('ZCARD', key)

        if count < limit then
            -- Add current request
            redis.call('ZADD', key, now, request_id)
            redis.call('EXPIRE', key, window * 2)
            return {1, limit - count - 1}  -- allowed, remaining
        else
            return {0, 0}  -- denied
        end
        """

        self.script_sha = self.redis.script_load(self.script)

    def allow_request(self) -> tuple[bool, int]:
        """Check if request is allowed

        Returns:
            (allowed, remaining)
        """
        now = time.time()
        request_id = str(uuid.uuid4())

        try:
            result = self.redis.evalsha(
                self.script_sha,
                1,
                self.key,
                self.limit,
                self.window_seconds,
                now,
                request_id
            )

            return bool(result[0]), int(result[1])

        except redis.exceptions.NoScriptError:
            self.script_sha = self.redis.script_load(self.script)
            return self.allow_request()

    def get_remaining(self) -> int:
        """Get remaining requests in window"""
        now = time.time()
        window_start = now - self.window_seconds

        # Clean old requests
        self.redis.zremrangebyscore(self.key, 0, window_start)

        # Count remaining
        count = self.redis.zcard(self.key)
        return max(0, self.limit - count)

    def get_oldest_request(self) -> float:
        """Get timestamp of oldest request in window"""
        now = time.time()
        window_start = now - self.window_seconds

        # Clean old requests
        self.redis.zremrangebyscore(self.key, 0, window_start)

        # Get oldest
        oldest = self.redis.zrange(self.key, 0, 0, withscores=True)
        if oldest:
            return float(oldest[0][1])
        return now

# Usage
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = DistributedSlidingWindow(redis_client, "user:123", limit=100, window_seconds=60)

allowed, remaining = limiter.allow_request()
if allowed:
    print(f"Request allowed. {remaining} remaining")
else:
    oldest = limiter.get_oldest_request()
    retry_after = int(oldest + limiter.window_seconds - time.time())
    print(f"Rate limited. Try in {retry_after} seconds")

Best For:

• Fair rate limiting
• Preventing boundary bursts
• High-accuracy requirements

Pros:

• No boundary bursts
• Smooth rate limiting
• Fair to all users
• Accurate limiting

Cons:

• Higher memory usage (stores timestamps)
• More complex implementation
• Slower than fixed window

Sliding Window Counter (Hybrid)

Description: Combines fixed window simplicity with sliding window accuracy. Uses weighted counts from current and previous windows.

Mathematical Model:

current_window_count = requests in current window
previous_window_count = requests in previous window

window_progress = (current_time % window_duration) / window_duration
estimated_count = (previous_window_count * (1 - window_progress)) + current_window_count

allowed = estimated_count < limit

Python Implementation:

import time
import threading
from collections import defaultdict

class SlidingWindowCounter:
    """Sliding window counter (hybrid approach)

    Approximates sliding window with fixed window simplicity.
    """

    def __init__(self, limit: int, window_seconds: int):
        """
        Args:
            limit: Maximum requests in window
            window_seconds: Window duration
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters = defaultdict(int)
        self._lock = threading.Lock()

    def _get_window_id(self, timestamp: float) -> int:
        """Get window ID for timestamp"""
        return int(timestamp // self.window_seconds)

    def _estimate_count(self, now: float) -> float:
        """Estimate request count in sliding window"""
        current_window = self._get_window_id(now)
        previous_window = current_window - 1

        # Get counts
        current_count = self.counters[current_window]
        previous_count = self.counters[previous_window]

        # Calculate weight for previous window
        window_progress = (now % self.window_seconds) / self.window_seconds
        previous_weight = 1.0 - window_progress

        # Weighted sum
        return (previous_count * previous_weight) + current_count

    def allow_request(self) -> bool:
        """Check if request is allowed

        Returns:
            True if under limit, False otherwise
        """
        with self._lock:
            now = time.time()
            estimated = self._estimate_count(now)

            if estimated < self.limit:
                current_window = self._get_window_id(now)
                self.counters[current_window] += 1

                # Clean old windows
                old_window = current_window - 2
                if old_window in self.counters:
                    del self.counters[old_window]

                return True

            return False

    def get_remaining(self) -> int:
        """Get remaining requests (approximate)"""
        with self._lock:
            now = time.time()
            estimated = self._estimate_count(now)
            return max(0, int(self.limit - estimated))

# Usage
limiter = SlidingWindowCounter(limit=100, window_seconds=60)

if limiter.allow_request():
    response = handle_request()
else:
    return error_response("Rate limited", 429)

Best For:

• Balance between accuracy and simplicity
• Lower memory than pure sliding window
• Better than fixed window for boundary cases

Pros:

• More accurate than fixed window
• Less memory than pure sliding window
• Good balance of trade-offs

Cons:

• Approximate (not exact)
• Still allows some boundary bursts (reduced)

Distributed Rate Limiting

Redis-Based Rate Limiting

Why Redis?

• Atomic operations (INCR, ZADD, Lua scripts)
• Fast in-memory storage
• Built-in expiration (TTL)
• Distributed consistency
• High availability (Redis Sentinel, Cluster)

Connection Patterns:

import redis
from redis.cluster import RedisCluster
from redis.sentinel import Sentinel

# Single Redis instance
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    socket_timeout=1.0,
    socket_connect_timeout=1.0,
    retry_on_timeout=True
)

# Redis Sentinel (high availability)
sentinel = Sentinel(
    [('sentinel1', 26379), ('sentinel2', 26379)],
    socket_timeout=1.0
)
redis_client = sentinel.master_for('mymaster', socket_timeout=1.0)

# Redis Cluster (sharding)
redis_client = RedisCluster(
    host='redis-cluster',
    port=6379,
    decode_responses=True
)

# Connection pooling
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=50,
    decode_responses=True
)
redis_client = redis.Redis(connection_pool=pool)

Lua Scripts for Atomicity:

import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Token bucket Lua script
token_bucket_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local tokens_needed = tonumber(ARGV[3])
local now = tonumber(ARGV[4])

local state = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(state[1]) or capacity
local last_refill = tonumber(state[2]) or now

-- Refill
local elapsed = now - last_refill
tokens = math.min(capacity, tokens + (elapsed * refill_rate))

-- Consume
if tokens >= tokens_needed then
    tokens = tokens - tokens_needed
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 3600)
    return {1, tokens}
else
    return {0, tokens}
end
"""

# Load script
script_sha = redis_client.script_load(token_bucket_script)

# Execute atomically
def check_rate_limit(user_id: str, capacity: int, refill_rate: float, tokens: int = 1):
    import time
    now = time.time()

    result = redis_client.evalsha(
        script_sha,
        1,  # number of keys
        f"rate_limit:{user_id}",  # KEYS[1]
        capacity,  # ARGV[1]
        refill_rate,  # ARGV[2]
        tokens,  # ARGV[3]
        now  # ARGV[4]
    )

    allowed = bool(result[0])
    remaining = float(result[1])

    return allowed, remaining

# Usage
allowed, remaining = check_rate_limit("user:123", capacity=100, refill_rate=10)

Error Handling:

import redis
import logging

logger = logging.getLogger(__name__)

def rate_limit_with_fallback(limiter_func, *args, **kwargs):
    """Rate limit with fallback on Redis failure

    Fail-open strategy: Allow requests if Redis unavailable
    """
    try:
        return limiter_func(*args, **kwargs)

    except redis.exceptions.ConnectionError as e:
        logger.error(f"Redis connection failed: {e}")
        return True  # Fail open (allow request)

    except redis.exceptions.TimeoutError as e:
        logger.error(f"Redis timeout: {e}")
        return True  # Fail open

    except redis.exceptions.RedisError as e:
        logger.error(f"Redis error: {e}")
        return True  # Fail open

    except Exception as e:
        logger.exception(f"Unexpected error in rate limiter: {e}")
        return True  # Fail open

# Usage
allowed = rate_limit_with_fallback(limiter.allow_request)
if allowed:
    response = handle_request()
else:
    return error_response("Rate limited", 429)

Multi-Tier Rate Limiting

Hierarchical Limits:

from typing import Dict, List
from dataclasses import dataclass

@dataclass
class RateLimit:
    """Rate limit configuration"""
    limit: int
    window_seconds: int
    name: str

class MultiTierRateLimiter:
    """Multi-tier rate limiter with multiple time windows"""

    def __init__(self, redis_client: redis.Redis, key: str,
                 limits: List[RateLimit]):
        """
        Args:
            redis_client: Redis connection
            key: Base key for rate limiting
            limits: List of rate limits to enforce
        """
        self.redis = redis_client
        self.key = key
        self.limits = limits
        self.limiters = []

        # Create limiter for each tier
        for limit_config in limits:
            limiter = DistributedFixedWindow(
                redis_client,
                f"{key}:{limit_config.name}",
                limit_config.limit,
                limit_config.window_seconds
            )
            self.limiters.append((limit_config, limiter))

    def allow_request(self) -> tuple[bool, Dict]:
        """Check all rate limit tiers

        Returns:
            (allowed, details)
        """
        details = {}

        for limit_config, limiter in self.limiters:
            allowed, remaining, reset_time = limiter.allow_request()

            details[limit_config.name] = {
                'limit': limit_config.limit,
                'remaining': remaining,
                'reset': reset_time,
                'window': limit_config.window_seconds
            }

            if not allowed:
                # First tier that fails
                return False, details

        return True, details

# Usage
limits = [
    RateLimit(limit=10, window_seconds=1, name='per_second'),
    RateLimit(limit=100, window_seconds=60, name='per_minute'),
    RateLimit(limit=1000, window_seconds=3600, name='per_hour'),
    RateLimit(limit=10000, window_seconds=86400, name='per_day'),
]

limiter = MultiTierRateLimiter(redis_client, "user:123", limits)

allowed, details = limiter.allow_request()
if allowed:
    # Add all tier details to headers
    for tier_name, tier_info in details.items():
        response.headers[f'X-RateLimit-{tier_name}-Limit'] = str(tier_info['limit'])
        response.headers[f'X-RateLimit-{tier_name}-Remaining'] = str(tier_info['remaining'])
else:
    # Find most restrictive tier
    return error_response("Rate limit exceeded", 429, details)

HTTP Headers and Standards

RFC 6585: Additional HTTP Status Codes

429 Too Many Requests (RFC 6585, Section 4):

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 60

{
  "error": "rate_limit_exceeded",
  "message": "Too many requests. Please retry after 60 seconds."
}

Specification:

• Status code 429 indicates client has sent too many requests
• SHOULD include Retry-After header indicating when to retry
• MAY include details about rate limiting in response body

Reference: RFC 6585

Rate Limit Headers

De facto standard headers (based on GitHub, Twitter, Stripe):

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 45
X-RateLimit-Reset: 1234567890

Header Definitions:

• X-RateLimit-Limit: Maximum requests allowed in time window
• X-RateLimit-Remaining: Requests remaining in current window
• X-RateLimit-Reset: Unix timestamp when limit resets

Retry-After Header (RFC 7231):

Retry-After: 60          # Seconds to wait
Retry-After: Wed, 21 Oct 2025 07:28:00 GMT  # Absolute time

Implementation:

from flask import Flask, jsonify, make_response
import time

app = Flask(__name__)

def add_rate_limit_headers(response, limit: int, remaining: int, reset: int):
    """Add standard rate limit headers"""
    response.headers['X-RateLimit-Limit'] = str(limit)
    response.headers['X-RateLimit-Remaining'] = str(remaining)
    response.headers['X-RateLimit-Reset'] = str(reset)
    return response

def rate_limit_exceeded_response(limit: int, reset: int):
    """Create 429 response with proper headers"""
    retry_after = reset - int(time.time())

    response = make_response(jsonify({
        'error': 'rate_limit_exceeded',
        'message': f'Too many requests. Please retry after {retry_after} seconds.',
        'retry_after': retry_after,
        'limit': limit,
        'reset': reset
    }), 429)

    response.headers['Retry-After'] = str(max(0, retry_after))
    response.headers['X-RateLimit-Limit'] = str(limit)
    response.headers['X-RateLimit-Remaining'] = '0'
    response.headers['X-RateLimit-Reset'] = str(reset)

    return response

@app.route('/api/resource')
def get_resource():
    user_id = get_user_id()
    limiter = DistributedFixedWindow(redis_client, user_id, limit=100, window_seconds=60)

    allowed, remaining, reset = limiter.allow_request()

    if not allowed:
        return rate_limit_exceeded_response(limiter.limit, reset)

    # Process request
    data = {'resource': 'data'}
    response = make_response(jsonify(data), 200)
    return add_rate_limit_headers(response, limiter.limit, remaining, reset)

IETF Draft: RateLimit Headers:

A more recent standard is being developed:

RateLimit-Limit: 100
RateLimit-Remaining: 45
RateLimit-Reset: 60

Reference: draft-ietf-httpapi-ratelimit-headers

Storage Backends

Redis

Best For: Distributed systems, high throughput, atomic operations

Setup:

# Install Redis
brew install redis  # macOS
apt-get install redis-server  # Ubuntu

# Start Redis
redis-server

# Connect
redis-cli

Python Client:

import redis

# Basic connection
client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

# Test connection
client.ping()  # Returns True if connected

Key Patterns:

# Fixed window
key = f"rate_limit:fixed:{user_id}:{window_id}"
redis.incr(key)
redis.expire(key, window_seconds * 2)

# Sliding window (sorted set)
key = f"rate_limit:sliding:{user_id}"
redis.zadd(key, {request_id: timestamp})
redis.zremrangebyscore(key, 0, window_start)
redis.expire(key, window_seconds)

# Token bucket (hash)
key = f"rate_limit:token:{user_id}"
redis.hmset(key, {'tokens': tokens, 'last_refill': timestamp})
redis.expire(key, 3600)

Memcached

Best For: Simple counters, low memory overhead

Setup:

# Install
brew install memcached  # macOS
apt-get install memcached  # Ubuntu

# Start
memcached -d -m 64 -p 11211

Python Client:

import pymemcache.client.base

client = pymemcache.client.base.Client(('localhost', 11211))

def fixed_window_memcached(user_id: str, limit: int, window_seconds: int) -> bool:
    """Fixed window rate limiting with Memcached"""
    now = int(time.time())
    window_id = now // window_seconds
    key = f"rate_limit:{user_id}:{window_id}"

    # Increment counter
    try:
        count = client.incr(key, 1)
    except pymemcache.exceptions.MemcacheError:
        # Key doesn't exist, create it
        client.set(key, 1, expire=window_seconds * 2)
        count = 1

    return count <= limit

In-Memory (Local)

Best For: Single-instance applications, no external dependencies

Implementation:

import time
import threading
from collections import defaultdict
from typing import Dict

class InMemoryRateLimiter:
    """Thread-safe in-memory rate limiter"""

    def __init__(self):
        self.limits: Dict[str, Dict] = defaultdict(dict)
        self._lock = threading.Lock()

    def check_fixed_window(self, key: str, limit: int, window_seconds: int) -> bool:
        """Fixed window rate limiting"""
        now = time.time()
        window_id = int(now // window_seconds)

        with self._lock:
            # Initialize if needed
            if window_id not in self.limits[key]:
                self.limits[key][window_id] = 0

            # Clean old windows
            old_windows = [wid for wid in self.limits[key] if wid < window_id]
            for wid in old_windows:
                del self.limits[key][wid]

            # Check limit
            if self.limits[key][window_id] < limit:
                self.limits[key][window_id] += 1
                return True

            return False

    def check_token_bucket(self, key: str, capacity: int, refill_rate: float) -> bool:
        """Token bucket rate limiting"""
        now = time.time()

        with self._lock:
            # Initialize if needed
            if key not in self.limits:
                self.limits[key] = {
                    'tokens': capacity,
                    'last_refill': now
                }

            # Refi

…(truncated)