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:

# 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:

# 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:

  1. Flexible parameter handling: The decorator extracts the Request object from either positional or keyword arguments, making it compatible with various FastAPI route signatures.

  2. 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).

  3. User-specific rate limiting: The rate limiting key uses the user's ID for authenticated users, which means each user gets their own quota.

  4. Graceful fallback: If token verification fails or no token is provided, the limiter falls back to IP-based limiting.

  5. 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:

# 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:

# 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:

Testing 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.

Previous
Next
Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal