Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/sentry_sdk/integrations/django/middleware.py: 78%

98 statements  

« prev     ^ index     » next       coverage.py v6.4.4, created at 2023-07-17 14:22 -0600

1""" 

2Create spans from Django middleware invocations 

3""" 

4 

5from django import VERSION as DJANGO_VERSION 

6 

7from sentry_sdk import Hub 

8from sentry_sdk._functools import wraps 

9from sentry_sdk._types import MYPY 

10from sentry_sdk.utils import ( 

11 ContextVar, 

12 transaction_from_function, 

13 capture_internal_exceptions, 

14) 

15 

16if MYPY: 16 ↛ 17line 16 didn't jump to line 17, because the condition on line 16 was never true

17 from typing import Any 

18 from typing import Callable 

19 from typing import Optional 

20 from typing import TypeVar 

21 

22 from sentry_sdk.tracing import Span 

23 

24 F = TypeVar("F", bound=Callable[..., Any]) 

25 

26_import_string_should_wrap_middleware = ContextVar( 

27 "import_string_should_wrap_middleware" 

28) 

29 

30if DJANGO_VERSION < (1, 7): 30 ↛ 31line 30 didn't jump to line 31, because the condition on line 30 was never true

31 import_string_name = "import_by_path" 

32else: 

33 import_string_name = "import_string" 

34 

35 

36if DJANGO_VERSION < (3, 1): 36 ↛ 37line 36 didn't jump to line 37, because the condition on line 36 was never true

37 _asgi_middleware_mixin_factory = lambda _: object 

38else: 

39 from .asgi import _asgi_middleware_mixin_factory 

40 

41 

42def patch_django_middlewares(): 

43 # type: () -> None 

44 from django.core.handlers import base 

45 

46 old_import_string = getattr(base, import_string_name) 

47 

48 def sentry_patched_import_string(dotted_path): 

49 # type: (str) -> Any 

50 rv = old_import_string(dotted_path) 

51 

52 if _import_string_should_wrap_middleware.get(None): 52 ↛ 55line 52 didn't jump to line 55, because the condition on line 52 was never false

53 rv = _wrap_middleware(rv, dotted_path) 

54 

55 return rv 

56 

57 setattr(base, import_string_name, sentry_patched_import_string) 

58 

59 old_load_middleware = base.BaseHandler.load_middleware 

60 

61 def sentry_patched_load_middleware(*args, **kwargs): 

62 # type: (Any, Any) -> Any 

63 _import_string_should_wrap_middleware.set(True) 

64 try: 

65 return old_load_middleware(*args, **kwargs) 

66 finally: 

67 _import_string_should_wrap_middleware.set(False) 

68 

69 base.BaseHandler.load_middleware = sentry_patched_load_middleware 

70 

71 

72def _wrap_middleware(middleware, middleware_name): 

73 # type: (Any, str) -> Any 

74 from sentry_sdk.integrations.django import DjangoIntegration 

75 

76 def _check_middleware_span(old_method): 

77 # type: (Callable[..., Any]) -> Optional[Span] 

78 hub = Hub.current 

79 integration = hub.get_integration(DjangoIntegration) 

80 if integration is None or not integration.middleware_spans: 80 ↛ 81line 80 didn't jump to line 81, because the condition on line 80 was never true

81 return None 

82 

83 function_name = transaction_from_function(old_method) 

84 

85 description = middleware_name 

86 function_basename = getattr(old_method, "__name__", None) 

87 if function_basename: 87 ↛ 90line 87 didn't jump to line 90, because the condition on line 87 was never false

88 description = "{}.{}".format(description, function_basename) 

89 

90 middleware_span = hub.start_span( 

91 op="django.middleware", description=description 

92 ) 

93 middleware_span.set_tag("django.function_name", function_name) 

94 middleware_span.set_tag("django.middleware_name", middleware_name) 

95 

96 return middleware_span 

97 

98 def _get_wrapped_method(old_method): 

99 # type: (F) -> F 

100 with capture_internal_exceptions(): 

101 

102 def sentry_wrapped_method(*args, **kwargs): 

103 # type: (*Any, **Any) -> Any 

104 middleware_span = _check_middleware_span(old_method) 

105 

106 if middleware_span is None: 106 ↛ 107line 106 didn't jump to line 107, because the condition on line 106 was never true

107 return old_method(*args, **kwargs) 

108 

109 with middleware_span: 

110 return old_method(*args, **kwargs) 

111 

112 try: 

113 # fails for __call__ of function on Python 2 (see py2.7-django-1.11) 

114 sentry_wrapped_method = wraps(old_method)(sentry_wrapped_method) 

115 

116 # Necessary for Django 3.1 

117 sentry_wrapped_method.__self__ = old_method.__self__ # type: ignore 

118 except Exception: 

119 pass 

120 

121 return sentry_wrapped_method # type: ignore 

122 

123 return old_method 

124 

125 class SentryWrappingMiddleware( 

126 _asgi_middleware_mixin_factory(_check_middleware_span) # type: ignore 

127 ): 

128 

129 async_capable = getattr(middleware, "async_capable", False) 

130 

131 def __init__(self, get_response=None, *args, **kwargs): 

132 # type: (Optional[Callable[..., Any]], *Any, **Any) -> None 

133 if get_response: 133 ↛ 136line 133 didn't jump to line 136, because the condition on line 133 was never false

134 self._inner = middleware(get_response, *args, **kwargs) 

135 else: 

136 self._inner = middleware(*args, **kwargs) 

137 self.get_response = get_response 

138 self._call_method = None 

139 if self.async_capable: 139 ↛ exitline 139 didn't return from function '__init__', because the condition on line 139 was never false

140 super(SentryWrappingMiddleware, self).__init__(get_response) 

141 

142 # We need correct behavior for `hasattr()`, which we can only determine 

143 # when we have an instance of the middleware we're wrapping. 

144 def __getattr__(self, method_name): 

145 # type: (str) -> Any 

146 if method_name not in ( 

147 "process_request", 

148 "process_view", 

149 "process_template_response", 

150 "process_response", 

151 "process_exception", 

152 ): 

153 raise AttributeError() 

154 

155 old_method = getattr(self._inner, method_name) 

156 rv = _get_wrapped_method(old_method) 

157 self.__dict__[method_name] = rv 

158 return rv 

159 

160 def __call__(self, *args, **kwargs): 

161 # type: (*Any, **Any) -> Any 

162 if hasattr(self, "async_route_check") and self.async_route_check(): 162 ↛ 163line 162 didn't jump to line 163, because the condition on line 162 was never true

163 return self.__acall__(*args, **kwargs) 

164 

165 f = self._call_method 

166 if f is None: 

167 self._call_method = f = self._inner.__call__ 

168 

169 middleware_span = _check_middleware_span(old_method=f) 

170 

171 if middleware_span is None: 171 ↛ 172line 171 didn't jump to line 172, because the condition on line 171 was never true

172 return f(*args, **kwargs) 

173 

174 with middleware_span: 

175 return f(*args, **kwargs) 

176 

177 for attr in ( 

178 "__name__", 

179 "__module__", 

180 "__qualname__", 

181 ): 

182 if hasattr(middleware, attr): 182 ↛ 177line 182 didn't jump to line 177, because the condition on line 182 was never false

183 setattr(SentryWrappingMiddleware, attr, getattr(middleware, attr)) 

184 

185 return SentryWrappingMiddleware