Snippet

Contextvar-Driven Error Tracking in Async ASGI Middleware

Async ASGI middleware uses contextvars to maintain thread-safe, coroutine-isolated request context across asynchronous execution flows when handling exceptions.

snippet.py

python

import contextvars
from django.core.exceptions import PermissionDenied
 
request_id_var = contextvars.ContextVar('request_id', default=None)
 
class AsyncExceptionContextMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
 
    async 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 exc
        finally:
            request_id_var.reset(token)

django

Breakdown

request_id_var = contextvars.ContextVar('request_id', default=None)

Instantiates a ContextVar object to safely pass state through asynchronous task chains.

token = request_id_var.set(request.headers.get('X-Request-ID', 'anon'))

Binds the incoming request header identifier to the current async context and returns a reset token.

finally: request_id_var.reset(token)

Ensures context isolation by cleaning up the ContextVar state after async execution completes.

Previous snippet Next snippet

More Expert Python exercises →

Contextvar-Driven Error Tracking in Async ASGI Middleware

Related