Protect APIs with rate limiting. Covers token bucket, sliding window, distributed rate limiting with Redis, client-specific limits, and graceful degradation under load.
This reference covers how to implement and manage rate limits and throttling for HTTP APIs, focusing on the core mechanisms, configuration options, and failure modes. The patterns described work across platforms that support standard HTTP headers and rate limiting protocols.
Rate limiting controls how many requests a client can make to an API within a specified time window. Throttling is the enforcement mechanism that applies these limits.
A basic rate limit consists of:
API responses include standard headers to communicate rate limits:
HTTP/1.1 200 OK
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 999
X-RateLimit-Reset: 1678886400
The X-RateLimit-Reset value is a Unix timestamp indicating when the limit resets.
Rate limits can be applied per:
Example configuration for an HTTP server:
# In application configuration
rate_limit:
window_seconds: 60
max_requests: 1000
identifier: "api_key" # or "ip_address"
Some APIs limit different resource types separately:
GET /api/users
HTTP/1.1 200 OK
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 999
X-RateLimit-Reset: 1678886400
X-RateLimit-Resource: "users"
When a client exceeds their rate limit:
HTTP/1.1 429 Too Many Requests
Retry-After: 60
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1678886400
The Retry-After header indicates how long to wait before retrying.
A simple in-memory implementation:
import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(list)
def is_allowed(self, identifier: str) -> bool:
now = time.time()
# Remove old requests outside the window
self.requests[identifier] = [
req_time for req_time in self.requests[identifier]
if now - req_time < self.window_seconds
]
if len(self.requests[identifier]) >= self.max_requests:
return False
self.requests[identifier].append(now)
return True
Common window durations include:
Example configuration with different window sizes:
# Per-minute limit
rate_limit:
window_seconds: 60
max_requests: 100
# Per-hour limit
rate_limit:
window_seconds: 3600
max_requests: 1000
Typical limits range from hundreds to tens of thousands per window.
The identifier used for rate limiting:
# By IP address
identifier: "ip_address"
# By API key
identifier: "api_key"
# By user ID
identifier: "user_id"
When a client hits their limit, they receive:
HTTP/1.1 429 Too Many Requests
Retry-After: 30
Content-Type: application/json
{
"error": "rate_limit_exceeded",
"message": "You have exceeded your rate limit"
}
Clients should implement exponential backoff:
import time
import requests
def make_request_with_retry(url, max_retries=3):
for attempt in range(max_retries):
try:
response = requests.get(url)
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 1))
time.sleep(retry_after)
continue
return response
except requests.RequestException:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
return response
A sliding window maintains request timestamps and calculates usage within the window:
def sliding_window_check(requests_list, window_seconds):
now = time.time()
valid_requests = [
req_time for req_time in requests_list
if now - req_time < window_seconds
]
return len(valid_requests)
A token bucket allows bursts up to a maximum:
import time
class TokenBucket:
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_refill = time.time()
def consume(self, tokens: int) -> bool:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
Different clients may have different limits:
# Tier 1: Free users (100 requests/minute)
# Tier 2: Premium users (1000 requests/minute)
# Tier 3: API partners (10000 requests/minute)
rate_limits:
- tier: "free"
window_seconds: 60
max_requests: 100
- tier: "premium"
window_seconds: 60
max_requests: 1000
- tier: "partner"
window_seconds: 60
max_requests: 10000
rate_limiting:
enabled: true
default_window_seconds: 60
default_max_requests: 1000
ip_limits:
window_seconds: 300
max_requests: 500
api_key_limits:
window_seconds: 60
max_requests: 1000
RATE_LIMIT_WINDOW_SECONDS=60
RATE_LIMIT_MAX_REQUESTS=1000
RATE_LIMIT_ENABLED=true
Rate limiting should be logged for observability:
{
"timestamp": "2023-03-15T14:30:00Z",
"client_id": "api_key_12345",
"request_path": "/api/users",
"method": "GET",
"status_code": 429,
"limit": 1000,
"remaining": 0,
"reset_time": 1678886400
}
Some APIs silently drop requests instead of returning 429:
# Without proper handling, a client might not know it was rate limited
response = requests.get(url)
# No error returned, but no data either
The X-RateLimit-Reset header may be in wrong format:
# Wrong: ISO timestamp instead of Unix timestamp
X-RateLimit-Reset: "2023-03-15T14:30:00Z"
# Correct: Unix timestamp
X-RateLimit-Reset: 1678886400
Using too short a window leads to frequent rate limit hits:
# Bad: 1-second window for high-volume API
rate_limit:
window_seconds: 1
max_requests: 1000
# Better: 60-second window
rate_limit:
window_seconds: 60
max_requests: 1000
Using the wrong identifier for client identification:
# Wrong: Using IP when API key is the real identifier
identifier: "ip_address"
# Correct: Using API key
identifier: "api_key"
Consistent error responses help clients handle rate limits properly:
HTTP/1.1 429 Too Many Requests
Content-Type: application/json
{
"error": "rate_limit_exceeded",
"message": "Rate limit exceeded",
"retry_after": 60,
"limit": 1000,
"remaining": 0,
"reset_time": 1678886400
}
Proper client-side handling:
import time
import requests
def safe_request(url, max_retries=3):
for attempt in range(max_retries):
response = requests.get(url)
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 1))
time.sleep(retry_after)
continue
return response
import unittest
class TestRateLimiter(unittest.TestCase):
def test_limit_exceeded(self):
limiter = RateLimiter(max_requests=2, window_seconds=60)
# Allow 2 requests
self.assertTrue(limiter.is_allowed("client1"))
self.assertTrue(limiter.is_allowed("client1"))
# Block the third request
self.assertFalse(limiter.is_allowed("client1"))
Testing full HTTP flow with rate limits:
import requests
def test_rate_limiting():
response = requests.get("http://api.example.com/limited-endpoint")
assert response.status_code == 429
assert 'Retry-After' in response.headers
In distributed deployments, rate limiting requires shared state:
# Using Redis for shared rate limit tracking
redis:
host: "localhost"
port: 6379
db: 0
When limits reset, ensure proper cache invalidation:
def reset_rate_limit(identifier):
# Clear the stored requests for this identifier
redis.delete(f"rate_limit:{identifier}")
Common default values:
default_window_seconds: 60
default_max_requests: 1000
Different endpoints may have different limits:
endpoints:
"/api/users":
window_seconds: 60
max_requests: 1000
"/api/admin":
window_seconds: 300
max_requests: 100
Race conditions in rate limiting implementations:
# Without proper synchronization, concurrent requests might bypass limits
class UnsafeRateLimiter:
def is_allowed(self, identifier):
# This might allow more requests than configured
# due to race conditions in checking and incrementing
pass
In-memory rate limiters grow with active clients:
# Memory usage grows with number of unique identifiers
# Consider using external storage for large-scale systems
Clock drift between servers affects rate limit calculations:
# Use NTP to synchronize clocks
# Ensure all servers use the same time source
Implement a request queue to handle rate limits gracefully:
import asyncio
import time
class RateLimitedClient:
def __init__(self, max_requests: int, window_seconds: int):
self.semaphore = asyncio.Semaphore(max_requests)
self.window_seconds = window_seconds
async def make_request(self, url):
async with self.semaphore:
# Make request
pass
Implement exponential backoff:
import time
import random
def exponential_backoff(attempt, base_delay=1, max_delay=60):
delay = min(base_delay * (2 ** attempt), max_delay)
return delay + random.uniform(0, 1) # Add jitter
Monitor for patterns in rate limiting events:
# Look for sudden spikes in 429 responses
# Check if certain clients hit limits more often
Use standard HTTP monitoring tools to track rate limit compliance.
The patterns described here work with any HTTP-based API system that implements standard headers and follows HTTP semantics. The specific implementation details depend on the underlying platform, but the core concepts remain consistent.
This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy