| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990 |
- 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: "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() -> 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: "Signal", sender: "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: "Callable[..., Any]",
- ) -> "Callable[..., Any]":
- @wraps(receiver)
- def wrapper(*args: "Any", **kwargs: "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
|
