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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
from asgiref.sync import iscoroutinefunction, markcoroutinefunction
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 iscoroutinefunction(self.get_response):
markcoroutinefunction(self)
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)
|
