import asyncio from django.http import Http404, HttpResponse from django.template import engines from django.template.response import TemplateResponse from django.utils.decorators import ( async_only_middleware, sync_and_async_middleware, sync_only_middleware, ) log = [] class BaseMiddleware: def __init__(self, get_response): self.get_response = get_response if asyncio.iscoroutinefunction(self.get_response): # Mark the class as async-capable. self._is_coroutine = asyncio.coroutines._is_coroutine def __call__(self, request): return self.get_response(request) class ProcessExceptionMiddleware(BaseMiddleware): def process_exception(self, request, exception): return HttpResponse("Exception caught") @async_only_middleware class AsyncProcessExceptionMiddleware(BaseMiddleware): async def process_exception(self, request, exception): return HttpResponse("Exception caught") class ProcessExceptionLogMiddleware(BaseMiddleware): def process_exception(self, request, exception): log.append("process-exception") class ProcessExceptionExcMiddleware(BaseMiddleware): def process_exception(self, request, exception): raise Exception("from process-exception") class ProcessViewMiddleware(BaseMiddleware): def process_view(self, request, view_func, view_args, view_kwargs): return HttpResponse("Processed view %s" % view_func.__name__) @async_only_middleware class AsyncProcessViewMiddleware(BaseMiddleware): async def process_view(self, request, view_func, view_args, view_kwargs): return HttpResponse("Processed view %s" % view_func.__name__) class ProcessViewNoneMiddleware(BaseMiddleware): def process_view(self, request, view_func, view_args, view_kwargs): log.append("processed view %s" % view_func.__name__) return None class ProcessViewTemplateResponseMiddleware(BaseMiddleware): def process_view(self, request, view_func, view_args, view_kwargs): template = engines["django"].from_string( "Processed view {{ view }}{% for m in mw %}\n{{ m }}{% endfor %}" ) return TemplateResponse( request, template, {"mw": [self.__class__.__name__], "view": view_func.__name__}, ) class TemplateResponseMiddleware(BaseMiddleware): def process_template_response(self, request, response): response.context_data["mw"].append(self.__class__.__name__) return response @async_only_middleware class AsyncTemplateResponseMiddleware(BaseMiddleware): async def process_template_response(self, request, response): response.context_data["mw"].append(self.__class__.__name__) return response class LogMiddleware(BaseMiddleware): def __call__(self, request): response = self.get_response(request) log.append((response.status_code, response.content)) return response class NoTemplateResponseMiddleware(BaseMiddleware): def process_template_response(self, request, response): return None @async_only_middleware class AsyncNoTemplateResponseMiddleware(BaseMiddleware): async def process_template_response(self, request, response): return None class NotFoundMiddleware(BaseMiddleware): def __call__(self, request): raise Http404("not found") class PaymentMiddleware(BaseMiddleware): def __call__(self, request): response = self.get_response(request) response.status_code = 402 return response @async_only_middleware def async_payment_middleware(get_response): async def middleware(request): response = await get_response(request) response.status_code = 402 return response return middleware @sync_and_async_middleware class SyncAndAsyncMiddleware(BaseMiddleware): pass @sync_only_middleware class DecoratedPaymentMiddleware(PaymentMiddleware): pass class NotSyncOrAsyncMiddleware(BaseMiddleware): """Middleware that is deliberately neither sync or async.""" sync_capable = False async_capable = False def __call__(self, request): return self.get_response(request)