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
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""
2Create spans from Django middleware invocations
3"""
5from django import VERSION as DJANGO_VERSION
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)
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
22 from sentry_sdk.tracing import Span
24 F = TypeVar("F", bound=Callable[..., Any])
26_import_string_should_wrap_middleware = ContextVar(
27 "import_string_should_wrap_middleware"
28)
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"
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
42def patch_django_middlewares():
43 # type: () -> None
44 from django.core.handlers import base
46 old_import_string = getattr(base, import_string_name)
48 def sentry_patched_import_string(dotted_path):
49 # type: (str) -> Any
50 rv = old_import_string(dotted_path)
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)
55 return rv
57 setattr(base, import_string_name, sentry_patched_import_string)
59 old_load_middleware = base.BaseHandler.load_middleware
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)
69 base.BaseHandler.load_middleware = sentry_patched_load_middleware
72def _wrap_middleware(middleware, middleware_name):
73 # type: (Any, str) -> Any
74 from sentry_sdk.integrations.django import DjangoIntegration
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
83 function_name = transaction_from_function(old_method)
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)
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)
96 return middleware_span
98 def _get_wrapped_method(old_method):
99 # type: (F) -> F
100 with capture_internal_exceptions():
102 def sentry_wrapped_method(*args, **kwargs):
103 # type: (*Any, **Any) -> Any
104 middleware_span = _check_middleware_span(old_method)
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)
109 with middleware_span:
110 return old_method(*args, **kwargs)
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)
116 # Necessary for Django 3.1
117 sentry_wrapped_method.__self__ = old_method.__self__ # type: ignore
118 except Exception:
119 pass
121 return sentry_wrapped_method # type: ignore
123 return old_method
125 class SentryWrappingMiddleware(
126 _asgi_middleware_mixin_factory(_check_middleware_span) # type: ignore
127 ):
129 async_capable = getattr(middleware, "async_capable", False)
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)
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()
155 old_method = getattr(self._inner, method_name)
156 rv = _get_wrapped_method(old_method)
157 self.__dict__[method_name] = rv
158 return rv
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)
165 f = self._call_method
166 if f is None:
167 self._call_method = f = self._inner.__call__
169 middleware_span = _check_middleware_span(old_method=f)
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)
174 with middleware_span:
175 return f(*args, **kwargs)
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))
185 return SentryWrappingMiddleware