python / expert
Snippet
Benutzerdefinierte Django Middleware mit Request/Response Verarbeitungspipeline
Django Middleware sitzen zwischen dem Webserver und Ihren Views und verarbeiten jede Anfrage und Antwort. Diese Experten-Level Middleware implementiert Rate-Limiting mit einem In-Memory-Zähler, verfolgt Anfragen pro IP-Adresse, blockiert übermäßige Anfragen mit HTTP 429 und injiziert benutzerdefinierte Header in Responses. Die __init__ erhält den get_response Callable, __call__ verarbeitet jedes Request/Response-Paar und process_exception behandelt unerwartete Fehler global.
snippet.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from django.http import JsonResponsefrom django.utils.deprecation import MiddlewareMixinclass RateLimitMiddleware(MiddlewareMixin):def __init__(self, get_response):self.get_response = get_responseself.request_counts = {}def __call__(self, request):client_ip = request.META.get('REMOTE_ADDR')self.request_counts[client_ip] = self.request_counts.get(client_ip, 0) + 1if self.request_counts[client_ip] > 100:return JsonResponse({'error': 'Rate limit exceeded'}, status=429)response = self.get_response(request)response['X-Request-Count'] = self.request_counts[client_ip]return responsedef process_exception(self, request, exception):return JsonResponse({'error': str(exception)}, status=500)
django
Erklärung
1
from django.utils.deprecation import MiddlewareMixin
Importiert MiddlewareMixin für Kompatibilität mit Djangos Middleware-Framework
2
def __init__(self, get_response):
Initialisierer erhält das nächste Middleware/View in der Kette als get_response
3
self.request_counts = {}
In-Memory Dictionary zum Speichern von Anfragenzählern pro Client-IP
4
client_ip = request.META.get('REMOTE_ADDR')
Extrahiert Client-IP aus den Request-Metadaten für Rate-Tracking
5
if self.request_counts[client_ip] > 100:
Prüft ob Client das Rate-Limit-Schwellenwert überschritten hat
6
return JsonResponse({'error': 'Rate limit exceeded'}, status=429)
Gibt 429 Too Many Requests zurück wenn Limit überschritten
7
response['X-Request-Count'] = self.request_counts[client_ip]
Injiziert benutzerdefinierten Header mit aktuellem Anfragenzähler in Response
8
def process_exception(self, request, exception):
Überschreiben um unbehandelte Ausnahmen global abzufangen und zu transformieren