Custom 429 Responses and Per User Limits

Introduction

Welcome to the third lesson of the "Implementing Rate Limiting" course! In our previous lessons, we explored global and endpoint-specific rate limiting. Now, we'll focus on enhancing the user experience by customizing 429 responses and implementing per-user rate limits to maintain a secure and user-friendly API.

Understanding HTTP 429 Status Code

The 429 status code is an HTTP response status code that indicates "Too Many Requests." It is used to signal that the user has sent too many requests in a given amount of time, exceeding the rate limit set by the server. Customizing these responses is important because it helps users understand why their requests are being blocked and what they can do next. By providing additional information, such as retry-after headers and links to support or documentation, we can guide users on how to proceed and improve their experience.

User Experience and Security Considerations

When implementing rate limiting, it's important to balance security with user experience. Without customized 429 responses, users may be confused about why their requests are being rejected. Similarly, without per-user rate limits, you might unfairly restrict legitimate users while failing to adequately limit malicious actors sharing IP addresses. By implementing user-specific rate limits, you can: Allow authenticated users higher request limits than anonymous users Prevent users on shared networks (like offices or universities) from being collectively penalized Block abusive users more effectively, even if they change IP addresses

Customizing 429 Responses

To improve the user experience, we first need to understand how to customize the 429 responses to provide clear and helpful messages. Let's look at a basic example of a custom rate limiter: Python# backend/middleware/rate_limiters.py from functools import wraps from fastapi import Request, HTTPException from collections import defaultdict import time import math # Store request counts per IP with timestamps request_store = defaultdict(list) def custom_limiter(func): """ Basic custom 429 response rate limiter - 5 requests per 30 seconds per IP """ @wraps(func) async def wrapper(request: Request, *args, **kwargs): current_time = time.time() window_seconds = 30 max_requests = 5 # Use IP as the key key = f"ip:{request.client.host}" # Clean old requests outside the window request_store[key] = [req_time for req_time in request_store[key] if current_time - req_time < window_seconds] # Check if limit exceeded if len(request_store[key]) >= max_requests: retry_after = int(window_seconds - (current_time - min(request_store[key]))) raise HTTPException( status_code=429, detail={ "error": "Too Many Requests", "message": "You have reached your request limit. Please try again later or contact support.", "retryAfter": retry_after } ) # Add current request request_store[key].append(current_time) # Call the original function return await func(request, *args, **kwargs) return wrapper In this code, we customize the 429 response by adding a retryAfter field, which calculates the time until the next request window. This approach not only informs users about the rate limit but also provides guidance on what to do next.

Implementing Per-User Rate Limits

Next, we'll implement a more sophisticated rate limiter that applies different limits based on whether the user is authenticated. Before we dive into the code, let's briefly explain what JWT is: JSON Web Tokens (JWT) are a compact, URL-safe means of representing claims securely between two parties. In web applications, they're commonly used for authentication - a server generates a token that certifies the user's identity, and the client includes this token with subsequent requests. This allows the server to verify who the user is without needing to store session data. Now, let's implement our user-based rate limiter: Python# backend/middleware/rate_limiters.py from functools import wraps from fastapi import Request, HTTPException from collections import defaultdict import time import jwt as jose_jwt from ..config import JWT_SECRET_KEY # Store request counts per user/IP with timestamps request_store = defaultdict(list) def user_based_limiter(func): """ User-based rate limiter decorator - Authenticated users: 10 requests per 30 seconds - Anonymous users: 5 requests per 30 seconds """ @wraps(func) async def wrapper(*args, **kwargs): # Extract request from args or kwargs request = None for arg in args: if isinstance(arg, Request): request = arg break if not request: request = kwargs.get('request') if not request: return await func(*args, **kwargs) current_time = time.time() window_seconds = 30 # Determine rate limit key and max requests auth_header = request.headers.get('authorization') if auth_header and auth_header.startswith('Bearer '): try: token = auth_header.split(' ')[1] decoded = jose_jwt.decode(token, JWT_SECRET_KEY, algorithms=['HS256']) key = f"user:{decoded['userId']}" max_requests = 10 # Authenticated users: 10 requests/30sec except: key = f"ip:{request.client.host}" max_requests = 5 # Invalid token, treat as anonymous else: key = f"ip:{request.client.host}" max_requests = 5 # Anonymous users: 5 requests/30sec # Clean old requests outside the window request_store[key] = [req_time for req_time in request_store[key] if current_time - req_time < window_seconds] # Check if limit exceeded if len(request_store[key]) >= max_requests: retry_after = int(window_seconds - (current_time - min(request_store[key]))) raise HTTPException( status_code=429, detail={ "error": "Too Many Requests", "message": "You have reached your request limit. Please try again later.", "retryAfter": retry_after } ) # Add current request request_store[key].append(current_time) # Call the original function return await func(*args, **kwargs) return wrapper This implementation has several key features: Flexible parameter handling: The decorator extracts the Request object from either positional or keyword arguments, making it compatible with various FastAPI route signatures. Dynamic request limits: The decorator checks if the user is authenticated by verifying their JWT token and applies different limits accordingly (10 requests for authenticated users, 5 for anonymous). User-specific rate limiting: The rate limiting key uses the user's ID for authenticated users, which means each user gets their own quota. Graceful fallback: If token verification fails or no token is provided, the limiter falls back to IP-based limiting. Informative error responses: When the rate limit is exceeded, users receive a clear message with a retryAfter field indicating when they can retry, along with guidance on what to do next. This approach is particularly effective for APIs that serve both authenticated and anonymous users.

Implementing in Routes

Now let's see how to apply these rate limiters to your API routes: Python# backend/routes/snippets.py from fastapi import APIRouter, Request from ..middleware.rate_limiters import custom_limiter, user_based_limiter router = APIRouter() # Route with custom 429 response @router.get("/public-endpoint") @custom_limiter async def public_endpoint(request: Request): return {"message": "Public endpoint with custom 429 responses"} # Route with user-based rate limiting @router.get("/test-rate-limit") @user_based_limiter async def test_rate_limit(request: Request): return {"message": "Rate limit test endpoint"} # Other route handlers... By adding the decorators to routes, we ensure that rate limiting is applied before the request handler is executed. The custom_limiter provides basic IP-based rate limiting with custom error messages, while the user_based_limiter offers more sophisticated per-user rate limiting that differentiates between authenticated and anonymous users.

Testing Per-User Rate Limiting

We can test our implementation using a script that checks both anonymous and authenticated access: Python# testRateLimiting.py import urllib.request import urllib.parse import urllib.error import json import jwt import time # Define a JWT secret key for testing purposes JWT_SECRET_KEY = 'jwt-secret-key' # Function to create a basic JWT token def create_token(user_id): return jwt.encode({'userId': user_id}, JWT_SECRET_KEY, algorithm='HS256') def main(): print("Testing per-user rate limits...") # Test anonymous user print("\nTesting as anonymous user:") for i in range(1, 8): try: with urllib.request.urlopen('http://localhost:3000/api/snippets/test-rate-limit') as response: print(f"Request {i}: {response.getcode()} Success") except urllib.error.HTTPError as e: try: error_response = e.read().decode() error_data = json.loads(error_response) print(f"Request {i}: {e.code} {error_data}") except: print(f"Request {i}: {e.code} {e.reason}") except Exception as e: print(f"Request {i}: Failed - {str(e)}") time.sleep(0.05) # Test authenticated user print("\nTesting as authenticated user:") token = create_token('user123') for i in range(1, 13): try: headers = {'Authorization': f'Bearer {token}'} req = urllib.request.Request('http://localhost:3000/api/snippets/test-rate-limit', headers=headers) with urllib.request.urlopen(req) as response: print(f"Request {i}: {response.getcode()} Success") except urllib.error.HTTPError as e: try: error_response = e.read().decode() error_data = json.loads(error_response) print(f"Request {i}: {e.code} {error_data}") except: print(f"Request {i}: {e.code} {e.reason}") except Exception as e: print(f"Request {i}: Failed - {str(e)}") time.sleep(0.05) if __name__ == "__main__": main() Output: plaintextTesting per-user rate limits... Testing as anonymous user: Request 1: 200 Success Request 2: 200 Success Request 3: 200 Success Request 4: 200 Success Request 5: 200 Success Request 6: 429 {'detail': {'error': 'Too Many Requests', 'message': 'You have reached your request limit. Please try again later.', 'retryAfter': 29}} Request 7: 429 {'detail': {'error': 'Too Many Requests', 'message': 'You have reached your request limit. Please try again later.', 'retryAfter': 29}} Testing as authenticated user: Request 1: 200 Success Request 2: 200 Success Request 3: 200 Success Request 4: 200 Success Request 5: 200 Success Request 6: 200 Success Request 7: 200 Success Request 8: 200 Success Request 9: 200 Success Request 10: 200 Success Request 11: 429 {'detail': {'error': 'Too Many Requests', 'message': 'You have reached your request limit. Please try again later.', 'retryAfter': 29}} Request 12: 429 {'detail': {'error': 'Too Many Requests', 'message': 'You have reached your request limit. Please try again later.', 'retryAfter': 29}}

Conclusion and Next Steps

You've now learned how to implement user-friendly 429 responses and per-user rate limits using Python and FastAPI. These techniques help create a secure and fair API while maintaining a positive user experience. In our next lesson, we'll build on this foundation to implement role-based rate limiting for more granular control over user access levels. In the upcoming practice, you'll apply the concepts from this lesson to reinforce your understanding.

Custom 429 Responses and Per User Limits

Introduction

Understanding HTTP 429 Status Code

User Experience and Security Considerations

Customizing 429 Responses

Implementing Per-User Rate Limits

Implementing in Routes

Testing Per-User Rate Limiting

Conclusion and Next Steps

Join the 1M+ learners on CodeSignal