opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
sentry_sdk
/
integrations
/
django
/
Go to Home Directory
+
Upload
Create File
root@0UT1S:~$
Execute
By Order of Mr.0UT1S
[DIR] ..
N/A
[DIR] __pycache__
N/A
__init__.py
24.38 KB
Rename
Delete
asgi.py
8.11 KB
Rename
Delete
caching.py
6.28 KB
Rename
Delete
middleware.py
5.87 KB
Rename
Delete
signals_handlers.py
3.03 KB
Rename
Delete
templates.py
5.59 KB
Rename
Delete
transactions.py
4.83 KB
Rename
Delete
views.py
3.06 KB
Rename
Delete
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
Save