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
|
# Python
import contextlib
import logging
import threading
_thread_locals = threading.local()
_logger = logging.getLogger('crum')
__version__ = '0.7.9'
__all__ = ['get_current_request', 'get_current_user', 'impersonate']
@contextlib.contextmanager
def impersonate(user=None):
"""
Temporarily impersonate the given user for audit trails.
"""
try:
current_user = get_current_user(_return_false=True)
set_current_user(user)
yield user
finally:
set_current_user(current_user)
def get_current_request():
"""
Return the request associated with the current thread.
"""
return getattr(_thread_locals, 'request', None)
def set_current_request(request=None):
"""
Update the request associated with the current thread.
"""
_thread_locals.request = request
# Clear the current user if also clearing the request.
if not request:
set_current_user(False)
def get_current_user(_return_false=False):
"""
Return the user associated with the current request thread.
"""
from crum.signals import current_user_getter
top_priority = -9999
top_user = False if _return_false else None
results = current_user_getter.send_robust(get_current_user)
for receiver, response in results:
priority = 0
if isinstance(response, Exception):
_logger.exception('%r raised exception: %s', receiver, response)
continue
elif isinstance(response, (tuple, list)) and response:
user = response[0]
if len(response) > 1:
priority = response[1]
elif response or response in (None, False):
user = response
else:
_logger.error('%r returned invalid response: %r', receiver,
response)
continue
if user is not False:
if priority > top_priority:
top_priority = priority
top_user = user
return top_user
def set_current_user(user=None):
"""
Update the user associated with the current request thread.
"""
from crum.signals import current_user_setter
results = current_user_setter.send_robust(set_current_user, user=user)
for receiver, response in results:
if isinstance(response, Exception):
_logger.exception('%r raised exception: %s', receiver, response)
class CurrentRequestUserMiddleware(object):
"""
Middleware to capture the request and user from the current thread.
"""
def __init__(self, get_response=None):
self.get_response = get_response
def __call__(self, request): # pragma: no cover
response = self.process_request(request)
try:
response = self.get_response(request)
except Exception as e:
self.process_exception(request, e)
raise
return self.process_response(request, response)
def process_request(self, request):
set_current_request(request)
def process_response(self, request, response):
set_current_request(None)
return response
def process_exception(self, request, exception):
set_current_request(None)
|
