python / expert
Snippet
Contextvar-gestütztes Fehler-Tracking in asynchroner ASGI-Middleware
Asynchrone ASGI-Middleware verwendet contextvars, um einen threadsicheren, Koroutinen-isolierten Anfrage-Kontext über asynchrone Ausführungsflüsse hinweg bei der Fehlerbehandlung aufrechtzuerhalten.
snippet.py
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import contextvarsfrom django.core.exceptions import PermissionDeniedrequest_id_var = contextvars.ContextVar('request_id', default=None)class AsyncExceptionContextMiddleware:def __init__(self, get_response):self.get_response = get_responseasync def __call__(self, request):token = request_id_var.set(request.headers.get('X-Request-ID', 'anon'))try:return await self.get_response(request)except PermissionDenied as exc:req_id = request_id_var.get()raise RuntimeError(f"[Req: {req_id}] Async Auth Failure: {exc}") from excfinally:request_id_var.reset(token)
django
Erklärung
1
request_id_var = contextvars.ContextVar('request_id', default=None)
Instanziiert ein ContextVar-Objekt zur sicheren Übergabe von Zuständen in asynchronen Task-Ketten.
2
token = request_id_var.set(request.headers.get('X-Request-ID', 'anon'))
Bindet die Header-ID der eingehenden Anfrage an den aktuellen asynchronen Kontext und gibt ein Reset-Token zurück.
3
finally: request_id_var.reset(token)
Stellt die Kontextisolierung sicher, indem der ContextVar-Zustand nach Abschluss der asynchronen Ausführung bereinigt wird.