Last Piece Hacktivist Crew

Editing: signals_handlers.py

from functools import wraps
from typing import TYPE_CHECKING

from django.dispatch import Signal

import sentry_sdk
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations.django import DJANGO_VERSION
from sentry_sdk.tracing_utils import has_span_streaming_enabled

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)

span_streaming = has_span_streaming_enabled(
                    sentry_sdk.get_client().options
                )
                if span_streaming:
                    with sentry_sdk.traces.start_span(
                        name=signal_name,
                        attributes={
                            "sentry.op": OP.EVENT_DJANGO,
                            "sentry.origin": DjangoIntegration.origin,
                            SPANDATA.CODE_FUNCTION_NAME: signal_name,
                        },
                    ) as span:
                        return receiver(*args, **kwargs)
                else:
                    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

Cancel