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
|
from functools import wraps
from django.dispatch import Signal
import sentry_sdk
from sentry_sdk.consts import OP
from sentry_sdk.integrations.django import DJANGO_VERSION
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
from typing import Any, Union
def _get_receiver_name(receiver):
# type: (Callable[..., Any]) -> str
name = ""
if hasattr(receiver, "__qualname__"):
name = receiver.__qualname__
elif hasattr(receiver, "__name__"): # Python 2.7 has no __qualname__
name = receiver.__name__
elif hasattr(
receiver, "func"
): # certain functions (like partials) dont have a name
if hasattr(receiver, "func") and hasattr(receiver.func, "__name__"):
name = "partial(<function " + receiver.func.__name__ + ">)"
if (
name == ""
): # In case nothing was found, return the string representation (this is the slowest case)
return str(receiver)
if hasattr(receiver, "__module__"): # prepend with module, if there is one
name = receiver.__module__ + "." + name
return name
def patch_signals():
# type: () -> None
"""
Patch django signal receivers to create a span.
This only wraps sync receivers. Django>=5.0 introduced async receivers, but
since we don't create transactions for ASGI Django, we don't wrap them.
"""
from sentry_sdk.integrations.django import DjangoIntegration
old_live_receivers = Signal._live_receivers
def _sentry_live_receivers(self, sender):
# type: (Signal, Any) -> Union[tuple[list[Callable[..., Any]], list[Callable[..., Any]]], list[Callable[..., Any]]]
if DJANGO_VERSION >= (5, 0):
sync_receivers, async_receivers = old_live_receivers(self, sender)
else:
sync_receivers = old_live_receivers(self, sender)
async_receivers = []
def sentry_sync_receiver_wrapper(receiver):
# type: (Callable[..., Any]) -> Callable[..., Any]
@wraps(receiver)
def wrapper(*args, **kwargs):
# type: (Any, Any) -> Any
signal_name = _get_receiver_name(receiver)
with sentry_sdk.start_span(
op=OP.EVENT_DJANGO,
name=signal_name,
origin=DjangoIntegration.origin,
) as span:
span.set_data("signal", signal_name)
return receiver(*args, **kwargs)
return wrapper
integration = sentry_sdk.get_client().get_integration(DjangoIntegration)
if (
integration
and integration.signals_spans
and self not in integration.signals_denylist
):
for idx, receiver in enumerate(sync_receivers):
sync_receivers[idx] = sentry_sync_receiver_wrapper(receiver)
if DJANGO_VERSION >= (5, 0):
return sync_receivers, async_receivers
else:
return sync_receivers
Signal._live_receivers = _sentry_live_receivers
|
