python / expert
Snippet
Django Custom Middleware for Request/Response Processing
Custom middleware in Django intercepts every request/response cycle, enabling cross-cutting concerns like rate limiting, request logging, and authentication preprocessing. This example demonstrates a rate-limiting middleware that tracks incoming requests per IP address within a sliding time window, returning 429 Too Many Requests when the limit is exceeded while adding rate limit headers to responses for client awareness.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import jsonfrom django.http import JsonResponsefrom django.utils.deprecation import MiddlewareMixinclass RateLimitMiddleware(MiddlewareMixin):"""Middleware to track and limit API requests per IP address."""request_counts = {}RATE_LIMIT = 100WINDOW_SECONDS = 60def __init__(self, get_response):self.get_response = get_responsesuper().__init__()def process_request(self, request):client_ip = request.META.get('HTTP_X_FORWARDED_FOR', '').split(',')[0] or \request.META.get('REMOTE_ADDR', '')import timecurrent_time = time.time()if client_ip in self.request_counts:timestamps = [t for t in self.request_counts[client_ip] if current_time - t < self.WINDOW_SECONDS]self.request_counts[client_ip] = timestampsif len(timestamps) >= self.RATE_LIMIT:return JsonResponse({'error': 'Rate limit exceeded'}, status=429)timestamps.append(current_time)else:self.request_counts[client_ip] = [current_time]request.client_ip = client_ipreturn Nonedef process_response(self, request, response):response['X-RateLimit-Limit'] = str(self.RATE_LIMIT)response['X-RateLimit-Remaining'] = str(self.RATE_LIMIT - len(self.request_counts.get(getattr(request, 'client_ip', ''), [])))return response
django
Breakdown
1
import json
JSON module for potential response serialization
2
from django.http import JsonResponse
Django response class for JSON data
3
from django.utils.deprecation import MiddlewareMixin
Deprecated class for backward-compatible middleware
4
class RateLimitMiddleware(MiddlewareMixin):
Custom middleware class inheriting from MiddlewareMixin
5
request_counts = {}
Class-level dictionary storing request timestamps per IP
6
def __init__(self, get_response):
Middleware initialization with response callable
7
client_ip = request.META.get('HTTP_X_FORWARDED_FOR', '').split(',')[0] or \
request.META.get('REMOTE_ADDR', '')
Extract client IP from headers, handling proxies
8
if len(timestamps) >= self.RATE_LIMIT:
Check if rate limit exceeded for this IP
9
return JsonResponse({'error': 'Rate limit exceeded'}, status=429)
Return JSON error response with 429 status code
10
response['X-RateLimit-Limit'] = str(self.RATE_LIMIT)
Add rate limit headers to outgoing response