python / expert
Snippet
Django Benutzerdefinierte Middleware für Request/Response-Verarbeitung
Benutzerdefinierte Middleware in Django fängt jeden Request/Response-Zyklus ab und ermöglicht übergreifende Anliegen wie Ratenbegrenzung, Request-Protokollierung und Authentifizierungsvorverarbeitung. Dieses Beispiel zeigt eine Rate-Limiting-Middleware, die eingehende Requests pro IP-Adresse in einem gleitenden Zeitfenster verfolgt, 429 Too Many Requests zurückgibt, wenn das Limit überschritten wird, und Rate-Limit-Header zu Responses für Client-Bewusstsein hinzufügt.
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
Erklärung
1
import json
JSON-Modul für mögliche Response-Serialisierung
2
from django.http import JsonResponse
Django Response-Klasse für JSON-Daten
3
from django.utils.deprecation import MiddlewareMixin
Veraltete Klasse für abwärtskompatible Middleware
4
class RateLimitMiddleware(MiddlewareMixin):
Benutzerdefinierte Middleware-Klasse erbt von MiddlewareMixin
5
request_counts = {}
Klassenvariables Wörterbuch speichert Request-Zeitstempel pro IP
6
def __init__(self, get_response):
Middleware-Initialisierung mit Response-Callable
7
client_ip = request.META.get('HTTP_X_FORWARDED_FOR', '').split(',')[0] or \
request.META.get('REMOTE_ADDR', '')
Extrahiere Client-IP aus Headern, behandle Proxies
8
if len(timestamps) >= self.RATE_LIMIT:
Prüfe ob Ratenlimit für diese IP überschritten
9
return JsonResponse({'error': 'Rate limit exceeded'}, status=429)
Gib JSON-Fehler-Response mit 429-Statuscode zurück
10
response['X-RateLimit-Limit'] = str(self.RATE_LIMIT)
Füge Rate-Limit-Header zu ausgehender Response hinzu