Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py # -- coding: utf-8 -- from __future__ import absolute_import import sys import threading import weakref from importlib import import_module from sentry_sdk._compat import string_types from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.hub import Hub, _should_send_default_pii from sentry_sdk.scope import add_global_event_processor from sentry_sdk.serializer import add_global_repr_processor from sentry_sdk.tracing import SOURCE_FOR_STYLE, TRANSACTION_SOURCE_URL from sentry_sdk.tracing_utils import record_sql_queries from sentry_sdk.utils import ( AnnotatedValue, HAS_REAL_CONTEXTVARS, CONTEXTVARS_ERROR_MESSAGE, SENSITIVE_DATA_SUBSTITUTE, logger, capture_internal_exceptions, event_from_exception, transaction_from_function, walk_exception_chain, ) from sentry_sdk.integrations import Integration, DidNotEnable from sentry_sdk.integrations.logging import ignore_logger from sentry_sdk.integrations.wsgi import SentryWsgiMiddleware from sentry_sdk.integrations._wsgi_common import RequestExtractor try: from django import VERSION as DJANGO_VERSION from django.conf import settings as django_settings from django.core import signals from django.conf import settings try: from django.urls import resolve except ImportError: from django.core.urlresolvers import resolve try: from django.urls import Resolver404 except ImportError: from django.core.urlresolvers import Resolver404 except ImportError: raise DidNotEnable("Django not installed") from sentry_sdk.integrations.django.transactions import LEGACY_RESOLVER from sentry_sdk.integrations.django.templates import ( get_template_frame_from_exception, patch_templates, ) from sentry_sdk.integrations.django.middleware import patch_django_middlewares from sentry_sdk.integrations.django.signals_handlers import patch_signals from sentry_sdk.integrations.django.views import patch_views if DJANGO_VERSION[:2] > (1, 8): from sentry_sdk.integrations.django.caching import patch_caching else: patch_caching = None # type: ignore if TYPE_CHECKING: from typing import Any from typing import Callable from typing import Dict from typing import Optional from typing import Union from typing import List from django.core.handlers.wsgi import WSGIRequest from django.http.response import HttpResponse from django.http.request import QueryDict from django.utils.datastructures import MultiValueDict from sentry_sdk.tracing import Span from sentry_sdk.scope import Scope from sentry_sdk.integrations.wsgi import _ScopedResponse from sentry_sdk._types import Event, Hint, EventProcessor, NotImplementedType if DJANGO_VERSION < (1, 10): def is_authenticated(request_user): # type: (Any) -> bool return request_user.is_authenticated() else: def is_authenticated(request_user): # type: (Any) -> bool return request_user.is_authenticated TRANSACTION_STYLE_VALUES = ("function_name", "url") class DjangoIntegration(Integration): identifier = "django" transaction_style = "" middleware_spans = None signals_spans = None cache_spans = None def __init__( self, transaction_style="url", middleware_spans=True, signals_spans=True, cache_spans=False, ): # type: (str, bool, bool, bool) -> None if transaction_style not in TRANSACTION_STYLE_VALUES: raise ValueError( "Invalid value for transaction_style: %s (must be in %s)" % (transaction_style, TRANSACTION_STYLE_VALUES) ) self.transaction_style = transaction_style self.middleware_spans = middleware_spans self.signals_spans = signals_spans self.cache_spans = cache_spans @staticmethod def setup_once(): # type: () -> None if DJANGO_VERSION < (1, 8): raise DidNotEnable("Django 1.8 or newer is required.") install_sql_hook() # Patch in our custom middleware. # logs an error for every 500 ignore_logger("django.server") ignore_logger("django.request") from django.core.handlers.wsgi import WSGIHandler old_app = WSGIHandler.__call__ def sentry_patched_wsgi_handler(self, environ, start_response): # type: (Any, Dict[str, str], Callable[..., Any]) -> _ScopedResponse if Hub.current.get_integration(DjangoIntegration) is None: return old_app(self, environ, start_response) bound_old_app = old_app.__get__(self, WSGIHandler) from django.conf import settings use_x_forwarded_for = settings.USE_X_FORWARDED_HOST return SentryWsgiMiddleware(bound_old_app, use_x_forwarded_for)( environ, start_response ) WSGIHandler.__call__ = sentry_patched_wsgi_handler _patch_get_response() _patch_django_asgi_handler() signals.got_request_exception.connect(_got_request_exception) @add_global_event_processor def process_django_templates(event, hint): # type: (Event, Optional[Hint]) -> Optional[Event] if hint is None: return event exc_info = hint.get("exc_info", None) if exc_info is None: return event exception = event.get("exception", None) if exception is None: return event values = exception.get("values", None) if values is None: return event for exception, (_, exc_value, _) in zip( reversed(values), walk_exception_chain(exc_info) ): frame = get_template_frame_from_exception(exc_value) if frame is not None: frames = exception.get("stacktrace", {}).get("frames", []) for i in reversed(range(len(frames))): f = frames[i] if ( f.get("function") in ("Parser.parse", "parse", "render") and f.get("module") == "django.template.base" ): i += 1 break else: i = len(frames) frames.insert(i, frame) return event @add_global_repr_processor def _django_queryset_repr(value, hint): # type: (Any, Dict[str, Any]) -> Union[NotImplementedType, str] try: # Django 1.6 can fail to import `QuerySet` when Django settings # have not yet been initialized. # # If we fail to import, return `NotImplemented`. It's at least # unlikely that we have a query set in `value` when importing # `QuerySet` fails. from django.db.models.query import QuerySet except Exception: return NotImplemented if not isinstance(value, QuerySet) or value._result_cache: return NotImplemented # Do not call Hub.get_integration here. It is intentional that # running under a new hub does not suddenly start executing # querysets. This might be surprising to the user but it's likely # less annoying. return "<%s from %s at 0x%x>" % ( value.__class__.__name__, value.__module__, id(value), ) _patch_channels() patch_django_middlewares() patch_views() patch_templates() patch_signals() if patch_caching is not None: patch_caching() _DRF_PATCHED = False _DRF_PATCH_LOCK = threading.Lock() def _patch_drf(): # type: () -> None """ Patch Django Rest Framework for more/better request data. DRF's request type is a wrapper around Django's request type. The attribute we're interested in is `request.data`, which is a cached property containing a parsed request body. Reading a request body from that property is more reliable than reading from any of Django's own properties, as those don't hold payloads in memory and therefore can only be accessed once. We patch the Django request object to include a weak backreference to the DRF request object, such that we can later use either in `DjangoRequestExtractor`. This function is not called directly on SDK setup, because importing almost any part of Django Rest Framework will try to access Django settings (where `sentry_sdk.init()` might be called from in the first place). Instead we run this function on every request and do the patching on the first request. """ global _DRF_PATCHED if _DRF_PATCHED: # Double-checked locking return with _DRF_PATCH_LOCK: if _DRF_PATCHED: return # We set this regardless of whether the code below succeeds or fails. # There is no point in trying to patch again on the next request. _DRF_PATCHED = True with capture_internal_exceptions(): try: from rest_framework.views import APIView # type: ignore except ImportError: pass else: old_drf_initial = APIView.initial def sentry_patched_drf_initial(self, request, args, kwargs): # type: (APIView, Any, Any, *Any) -> Any with capture_internal_exceptions(): request._request._sentry_drf_request_backref = weakref.ref( request ) pass return old_drf_initial(self, request, args, kwargs) APIView.initial = sentry_patched_drf_initial def _patch_channels(): # type: () -> None try: from channels.http import AsgiHandler # type: ignore except ImportError: return if not HAS_REAL_CONTEXTVARS: # We better have contextvars or we're going to leak state between # requests. # # We cannot hard-raise here because channels may not be used at all in # the current process. That is the case when running traditional WSGI # workers in gunicorn+gevent and the websocket stuff in a separate # process. logger.warning( "We detected that you are using Django channels 2.0." + CONTEXTVARS_ERROR_MESSAGE ) from sentry_sdk.integrations.django.asgi import patch_channels_asgi_handler_impl patch_channels_asgi_handler_impl(AsgiHandler) def _patch_django_asgi_handler(): # type: () -> None try: from django.core.handlers.asgi import ASGIHandler except ImportError: return if not HAS_REAL_CONTEXTVARS: # We better have contextvars or we're going to leak state between # requests. # # We cannot hard-raise here because Django's ASGI stuff may not be used # at all. logger.warning( "We detected that you are using Django 3." + CONTEXTVARS_ERROR_MESSAGE ) from sentry_sdk.integrations.django.asgi import patch_django_asgi_handler_impl patch_django_asgi_handler_impl(ASGIHandler) def _set_transaction_name_and_source(scope, transaction_style, request): # type: (Scope, str, WSGIRequest) -> None try: transaction_name = None if transaction_style == "function_name": fn = resolve(request.path).func transaction_name = transaction_from_function(getattr(fn, "view_class", fn)) elif transaction_style == "url": if hasattr(request, "urlconf"): transaction_name = LEGACY_RESOLVER.resolve( request.path_info, urlconf=request.urlconf ) else: transaction_name = LEGACY_RESOLVER.resolve(request.path_info) if transaction_name is None: transaction_name = request.path_info source = TRANSACTION_SOURCE_URL else: source = SOURCE_FOR_STYLE[transaction_style] scope.set_transaction_name( transaction_name, source=source, ) except Resolver404: urlconf = import_module(settings.ROOT_URLCONF) # This exception only gets thrown when transaction_style is `function_name` # So we don't check here what style is configured if hasattr(urlconf, "handler404"): handler = urlconf.handler404 if isinstance(handler, string_types): scope.transaction = handler else: scope.transaction = transaction_from_function( getattr(handler, "view_class", handler) ) except Exception: pass def _before_get_response(request): # type: (WSGIRequest) -> None hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is None: return _patch_drf() with hub.configure_scope() as scope: # Rely on WSGI middleware to start a trace _set_transaction_name_and_source(scope, integration.transaction_style, request) scope.add_event_processor( _make_event_processor(weakref.ref(request), integration) ) def _attempt_resolve_again(request, scope, transaction_style): # type: (WSGIRequest, Scope, str) -> None """ Some django middlewares overwrite request.urlconf so we need to respect that contract, so we try to resolve the url again. """ if not hasattr(request, "urlconf"): return _set_transaction_name_and_source(scope, transaction_style, request) def _after_get_response(request): # type: (WSGIRequest) -> None hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is None or integration.transaction_style != "url": return with hub.configure_scope() as scope: _attempt_resolve_again(request, scope, integration.transaction_style) def _patch_get_response(): # type: () -> None """ patch get_response, because at that point we have the Django request object """ from django.core.handlers.base import BaseHandler old_get_response = BaseHandler.get_response def sentry_patched_get_response(self, request): # type: (Any, WSGIRequest) -> Union[HttpResponse, BaseException] _before_get_response(request) rv = old_get_response(self, request) _after_get_response(request) return rv BaseHandler.get_response = sentry_patched_get_response if hasattr(BaseHandler, "get_response_async"): from sentry_sdk.integrations.django.asgi import patch_get_response_async patch_get_response_async(BaseHandler, _before_get_response) def _make_event_processor(weak_request, integration): # type: (Callable[[], WSGIRequest], DjangoIntegration) -> EventProcessor def event_processor(event, hint): # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any] # if the request is gone we are fine not logging the data from # it. This might happen if the processor is pushed away to # another thread. request = weak_request() if request is None: return event try: drf_request = request._sentry_drf_request_backref() if drf_request is not None: request = drf_request except AttributeError: pass with capture_internal_exceptions(): DjangoRequestExtractor(request).extract_into_event(event) if _should_send_default_pii(): with capture_internal_exceptions(): _set_user_info(request, event) return event return event_processor def _got_request_exception(request=None, kwargs): # type: (WSGIRequest, **Any) -> None hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is not None: if request is not None and integration.transaction_style == "url": with hub.configure_scope() as scope: _attempt_resolve_again(request, scope, integration.transaction_style) # If an integration is there, a client has to be there. client = hub.client # type: Any event, hint = event_from_exception( sys.exc_info(), client_options=client.options, mechanism={"type": "django", "handled": False}, ) hub.capture_event(event, hint=hint) class DjangoRequestExtractor(RequestExtractor): def env(self): # type: () -> Dict[str, str] return self.request.META def cookies(self): # type: () -> Dict[str, Union[str, AnnotatedValue]] privacy_cookies = [ django_settings.CSRF_COOKIE_NAME, django_settings.SESSION_COOKIE_NAME, ] clean_cookies = {} # type: Dict[str, Union[str, AnnotatedValue]] for key, val in self.request.COOKIES.items(): if key in privacy_cookies: clean_cookies[key] = SENSITIVE_DATA_SUBSTITUTE else: clean_cookies[key] = val return clean_cookies def raw_data(self): # type: () -> bytes return self.request.body def form(self): # type: () -> QueryDict return self.request.POST def files(self): # type: () -> MultiValueDict return self.request.FILES def size_of_file(self, file): # type: (Any) -> int return file.size def parsed_body(self): # type: () -> Optional[Dict[str, Any]] try: return self.request.data except AttributeError: return RequestExtractor.parsed_body(self) def _set_user_info(request, event): # type: (WSGIRequest, Dict[str, Any]) -> None user_info = event.setdefault("user", {}) user = getattr(request, "user", None) if user is None or not is_authenticated(user): return try: user_info.setdefault("id", str(user.pk)) except Exception: pass try: user_info.setdefault("email", user.email) except Exception: pass try: user_info.setdefault("username", user.get_username()) except Exception: pass def install_sql_hook(): # type: () -> None """If installed this causes Django's queries to be captured.""" try: from django.db.backends.utils import CursorWrapper except ImportError: from django.db.backends.util import CursorWrapper try: # django 1.6 and 1.7 compatability from django.db.backends import BaseDatabaseWrapper except ImportError: # django 1.8 or later from django.db.backends.base.base import BaseDatabaseWrapper try: real_execute = CursorWrapper.execute real_executemany = CursorWrapper.executemany real_connect = BaseDatabaseWrapper.connect except AttributeError: # This won't work on Django versions < 1.6 return def execute(self, sql, params=None): # type: (CursorWrapper, Any, Optional[Any]) -> Any hub = Hub.current if hub.get_integration(DjangoIntegration) is None: return real_execute(self, sql, params) with record_sql_queries( hub, self.cursor, sql, params, paramstyle="format", executemany=False ) as span: _set_db_data(span, self.db.vendor, self.db.get_connection_params()) return real_execute(self, sql, params) def executemany(self, sql, param_list): # type: (CursorWrapper, Any, List[Any]) -> Any hub = Hub.current if hub.get_integration(DjangoIntegration) is None: return real_executemany(self, sql, param_list) with record_sql_queries( hub, self.cursor, sql, param_list, paramstyle="format", executemany=True ) as span: _set_db_data(span, self.db.vendor, self.db.get_connection_params()) return real_executemany(self, sql, param_list) def connect(self): # type: (BaseDatabaseWrapper) -> None hub = Hub.current if hub.get_integration(DjangoIntegration) is None: return real_connect(self) with capture_internal_exceptions(): hub.add_breadcrumb(message="connect", category="query") with hub.start_span(op=OP.DB, description="connect") as span: _set_db_data(span, self.vendor, self.get_connection_params()) return real_connect(self) CursorWrapper.execute = execute CursorWrapper.executemany = executemany BaseDatabaseWrapper.connect = connect ignore_logger("django.db.backends") def _set_db_data(span, vendor, connection_params): # type: (Span, str, Dict[str, str]) -> None span.set_data(SPANDATA.DB_SYSTEM, vendor) db_name = connection_params.get("dbname") or connection_params.get("database") if db_name is not None: span.set_data(SPANDATA.DB_NAME, db_name) server_address = connection_params.get("host") if server_address is not None: span.set_data(SPANDATA.SERVER_ADDRESS, server_address) server_port = connection_params.get("port") if server_port is not None: span.set_data(SPANDATA.SERVER_PORT, server_port) server_socket_address = connection_params.get("unix_socket") if server_socket_address is not None: span.set_data(SPANDATA.SERVER_SOCKET_ADDRESS, server_socket_address)	21.4 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 asgi.py """ Instrumentation for Django 3.0 Since this file contains `async def` it is conditionally imported in `sentry_sdk.integrations.django` (depending on the existence of `django.core.handlers.asgi`. """ import asyncio from sentry_sdk import Hub, _functools from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP from sentry_sdk.integrations.asgi import SentryAsgiMiddleware if TYPE_CHECKING: from typing import Any from typing import Union from typing import Callable from django.http.response import HttpResponse def patch_django_asgi_handler_impl(cls): # type: (Any) -> None from sentry_sdk.integrations.django import DjangoIntegration old_app = cls.__call__ async def sentry_patched_asgi_handler(self, scope, receive, send): # type: (Any, Any, Any, Any) -> Any if Hub.current.get_integration(DjangoIntegration) is None: return await old_app(self, scope, receive, send) middleware = SentryAsgiMiddleware( old_app.__get__(self, cls), unsafe_context_data=True )._run_asgi3 return await middleware(scope, receive, send) cls.__call__ = sentry_patched_asgi_handler def patch_get_response_async(cls, _before_get_response): # type: (Any, Any) -> None old_get_response_async = cls.get_response_async async def sentry_patched_get_response_async(self, request): # type: (Any, Any) -> Union[HttpResponse, BaseException] _before_get_response(request) return await old_get_response_async(self, request) cls.get_response_async = sentry_patched_get_response_async def patch_channels_asgi_handler_impl(cls): # type: (Any) -> None import channels # type: ignore from sentry_sdk.integrations.django import DjangoIntegration if channels.__version__ < "3.0.0": old_app = cls.__call__ async def sentry_patched_asgi_handler(self, receive, send): # type: (Any, Any, Any) -> Any if Hub.current.get_integration(DjangoIntegration) is None: return await old_app(self, receive, send) middleware = SentryAsgiMiddleware( lambda _scope: old_app.__get__(self, cls), unsafe_context_data=True ) return await middleware(self.scope)(receive, send) cls.__call__ = sentry_patched_asgi_handler else: # The ASGI handler in Channels >= 3 has the same signature as # the Django handler. patch_django_asgi_handler_impl(cls) def wrap_async_view(hub, callback): # type: (Hub, Any) -> Any @_functools.wraps(callback) async def sentry_wrapped_callback(request, args, kwargs): # type: (Any, Any, *Any) -> Any with hub.configure_scope() as sentry_scope: if sentry_scope.profile is not None: sentry_scope.profile.update_active_thread_id() with hub.start_span( op=OP.VIEW_RENDER, description=request.resolver_match.view_name ): return await callback(request, args, *kwargs) return sentry_wrapped_callback def _asgi_middleware_mixin_factory(_check_middleware_span): # type: (Callable[..., Any]) -> Any """ Mixin class factory that generates a middleware mixin for handling requests in async mode. """ class SentryASGIMixin: if TYPE_CHECKING: _inner = None def __init__(self, get_response): # type: (Callable[..., Any]) -> None self.get_response = get_response self._acall_method = None self._async_check() def _async_check(self): # type: () -> None """ If get_response is a coroutine function, turns us into async mode so a thread is not consumed during a whole request. Taken from django.utils.deprecation::MiddlewareMixin._async_check """ if asyncio.iscoroutinefunction(self.get_response): self._is_coroutine = asyncio.coroutines._is_coroutine # type: ignore def async_route_check(self): # type: () -> bool """ Function that checks if we are in async mode, and if we are forwards the handling of requests to __acall__ """ return asyncio.iscoroutinefunction(self.get_response) async def __acall__(self, args, *kwargs): # type: (Any, *Any) -> Any f = self._acall_method if f is None: if hasattr(self._inner, "__acall__"): self._acall_method = f = self._inner.__acall__ # type: ignore else: self._acall_method = f = self._inner middleware_span = _check_middleware_span(old_method=f) if middleware_span is None: return await f(args, *kwargs) with middleware_span: return await f(args, **kwargs) return SentryASGIMixin	4.92 KB	2026-01-20 12:58
📄 caching.py import functools from typing import TYPE_CHECKING from django import VERSION as DJANGO_VERSION from django.core.cache import CacheHandler from sentry_sdk import Hub from sentry_sdk.consts import OP, SPANDATA from sentry_sdk._compat import text_type if TYPE_CHECKING: from typing import Any from typing import Callable METHODS_TO_INSTRUMENT = [ "get", "get_many", ] def _get_span_description(method_name, args, kwargs): # type: (str, Any, Any) -> str description = "{} ".format(method_name) if args is not None and len(args) >= 1: description += text_type(args[0]) elif kwargs is not None and "key" in kwargs: description += text_type(kwargs["key"]) return description def _patch_cache_method(cache, method_name): # type: (CacheHandler, str) -> None from sentry_sdk.integrations.django import DjangoIntegration def _instrument_call(cache, method_name, original_method, args, kwargs): # type: (CacheHandler, str, Callable[..., Any], Any, Any) -> Any hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is None or not integration.cache_spans: return original_method(args, kwargs) description = _get_span_description(method_name, args, kwargs) with hub.start_span(op=OP.CACHE_GET_ITEM, description=description) as span: value = original_method(args, *kwargs) if value: span.set_data(SPANDATA.CACHE_HIT, True) size = len(text_type(value)) span.set_data(SPANDATA.CACHE_ITEM_SIZE, size) else: span.set_data(SPANDATA.CACHE_HIT, False) return value original_method = getattr(cache, method_name) @functools.wraps(original_method) def sentry_method(args, *kwargs): # type: (Any, **Any) -> Any return _instrument_call(cache, method_name, original_method, args, kwargs) setattr(cache, method_name, sentry_method) def _patch_cache(cache): # type: (CacheHandler) -> None if not hasattr(cache, "_sentry_patched"): for method_name in METHODS_TO_INSTRUMENT: _patch_cache_method(cache, method_name) cache._sentry_patched = True def patch_caching(): # type: () -> None from sentry_sdk.integrations.django import DjangoIntegration if not hasattr(CacheHandler, "_sentry_patched"): if DJANGO_VERSION < (3, 2): original_get_item = CacheHandler.__getitem__ @functools.wraps(original_get_item) def sentry_get_item(self, alias): # type: (CacheHandler, str) -> Any cache = original_get_item(self, alias) integration = Hub.current.get_integration(DjangoIntegration) if integration and integration.cache_spans: _patch_cache(cache) return cache CacheHandler.__getitem__ = sentry_get_item CacheHandler._sentry_patched = True else: original_create_connection = CacheHandler.create_connection @functools.wraps(original_create_connection) def sentry_create_connection(self, alias): # type: (CacheHandler, str) -> Any cache = original_create_connection(self, alias) integration = Hub.current.get_integration(DjangoIntegration) if integration and integration.cache_spans: _patch_cache(cache) return cache CacheHandler.create_connection = sentry_create_connection CacheHandler._sentry_patched = True	3.61 KB	2026-01-20 12:58
📄 middleware.py """ Create spans from Django middleware invocations """ from django import VERSION as DJANGO_VERSION from sentry_sdk import Hub from sentry_sdk._functools import wraps from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP from sentry_sdk.utils import ( ContextVar, transaction_from_function, capture_internal_exceptions, ) if TYPE_CHECKING: from typing import Any from typing import Callable from typing import Optional from typing import TypeVar from sentry_sdk.tracing import Span F = TypeVar("F", bound=Callable[..., Any]) _import_string_should_wrap_middleware = ContextVar( "import_string_should_wrap_middleware" ) if DJANGO_VERSION < (1, 7): import_string_name = "import_by_path" else: import_string_name = "import_string" if DJANGO_VERSION < (3, 1): _asgi_middleware_mixin_factory = lambda _: object else: from .asgi import _asgi_middleware_mixin_factory def patch_django_middlewares(): # type: () -> None from django.core.handlers import base old_import_string = getattr(base, import_string_name) def sentry_patched_import_string(dotted_path): # type: (str) -> Any rv = old_import_string(dotted_path) if _import_string_should_wrap_middleware.get(None): rv = _wrap_middleware(rv, dotted_path) return rv setattr(base, import_string_name, sentry_patched_import_string) old_load_middleware = base.BaseHandler.load_middleware def sentry_patched_load_middleware(args, kwargs): # type: (Any, Any) -> Any _import_string_should_wrap_middleware.set(True) try: return old_load_middleware(args, *kwargs) finally: _import_string_should_wrap_middleware.set(False) base.BaseHandler.load_middleware = sentry_patched_load_middleware def _wrap_middleware(middleware, middleware_name): # type: (Any, str) -> Any from sentry_sdk.integrations.django import DjangoIntegration def _check_middleware_span(old_method): # type: (Callable[..., Any]) -> Optional[Span] hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is None or not integration.middleware_spans: return None function_name = transaction_from_function(old_method) description = middleware_name function_basename = getattr(old_method, "__name__", None) if function_basename: description = "{}.{}".format(description, function_basename) middleware_span = hub.start_span( op=OP.MIDDLEWARE_DJANGO, description=description ) middleware_span.set_tag("django.function_name", function_name) middleware_span.set_tag("django.middleware_name", middleware_name) return middleware_span def _get_wrapped_method(old_method): # type: (F) -> F with capture_internal_exceptions(): def sentry_wrapped_method(args, *kwargs): # type: (Any, *Any) -> Any middleware_span = _check_middleware_span(old_method) if middleware_span is None: return old_method(args, *kwargs) with middleware_span: return old_method(args, *kwargs) try: # fails for __call__ of function on Python 2 (see py2.7-django-1.11) sentry_wrapped_method = wraps(old_method)(sentry_wrapped_method) # Necessary for Django 3.1 sentry_wrapped_method.__self__ = old_method.__self__ # type: ignore except Exception: pass return sentry_wrapped_method # type: ignore return old_method class SentryWrappingMiddleware( _asgi_middleware_mixin_factory(_check_middleware_span) # type: ignore ): async_capable = getattr(middleware, "async_capable", False) def __init__(self, get_response=None, args, *kwargs): # type: (Optional[Callable[..., Any]], Any, *Any) -> None if get_response: self._inner = middleware(get_response, args, *kwargs) else: self._inner = middleware(args, *kwargs) self.get_response = get_response self._call_method = None if self.async_capable: super(SentryWrappingMiddleware, self).__init__(get_response) # We need correct behavior for `hasattr()`, which we can only determine # when we have an instance of the middleware we're wrapping. def __getattr__(self, method_name): # type: (str) -> Any if method_name not in ( "process_request", "process_view", "process_template_response", "process_response", "process_exception", ): raise AttributeError() old_method = getattr(self._inner, method_name) rv = _get_wrapped_method(old_method) self.__dict__[method_name] = rv return rv def __call__(self, args, *kwargs): # type: (Any, *Any) -> Any if hasattr(self, "async_route_check") and self.async_route_check(): return self.__acall__(args, *kwargs) f = self._call_method if f is None: self._call_method = f = self._inner.__call__ middleware_span = _check_middleware_span(old_method=f) if middleware_span is None: return f(args, *kwargs) with middleware_span: return f(args, **kwargs) for attr in ( "__name__", "__module__", "__qualname__", ): if hasattr(middleware, attr): setattr(SentryWrappingMiddleware, attr, getattr(middleware, attr)) return SentryWrappingMiddleware	5.83 KB	2026-01-20 12:58
📄 signals_handlers.py # -- coding: utf-8 -- from __future__ import absolute_import from django.dispatch import Signal from sentry_sdk import Hub from sentry_sdk._functools import wraps from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP if TYPE_CHECKING: from typing import Any from typing import Callable from typing import List 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""" from sentry_sdk.integrations.django import DjangoIntegration old_live_receivers = Signal._live_receivers def _sentry_live_receivers(self, sender): # type: (Signal, Any) -> List[Callable[..., Any]] hub = Hub.current receivers = old_live_receivers(self, sender) def sentry_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 hub.start_span( op=OP.EVENT_DJANGO, description=signal_name, ) as span: span.set_data("signal", signal_name) return receiver(args, **kwargs) return wrapper integration = hub.get_integration(DjangoIntegration) if integration and integration.signals_spans: for idx, receiver in enumerate(receivers): receivers[idx] = sentry_receiver_wrapper(receiver) return receivers Signal._live_receivers = _sentry_live_receivers	2.41 KB	2026-01-20 12:58
📄 templates.py from django.template import TemplateSyntaxError from django.utils.safestring import mark_safe from django import VERSION as DJANGO_VERSION from sentry_sdk import _functools, Hub from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP if TYPE_CHECKING: from typing import Any from typing import Dict from typing import Optional from typing import Iterator from typing import Tuple try: # support Django 1.9 from django.template.base import Origin except ImportError: # backward compatibility from django.template.loader import LoaderOrigin as Origin def get_template_frame_from_exception(exc_value): # type: (Optional[BaseException]) -> Optional[Dict[str, Any]] # As of Django 1.9 or so the new template debug thing showed up. if hasattr(exc_value, "template_debug"): return _get_template_frame_from_debug(exc_value.template_debug) # type: ignore # As of r16833 (Django) all exceptions may contain a # ``django_template_source`` attribute (rather than the legacy # ``TemplateSyntaxError.source`` check) if hasattr(exc_value, "django_template_source"): return _get_template_frame_from_source( exc_value.django_template_source # type: ignore ) if isinstance(exc_value, TemplateSyntaxError) and hasattr(exc_value, "source"): source = exc_value.source if isinstance(source, (tuple, list)) and isinstance(source[0], Origin): return _get_template_frame_from_source(source) # type: ignore return None def _get_template_name_description(template_name): # type: (str) -> str if isinstance(template_name, (list, tuple)): if template_name: return "[{}, ...]".format(template_name[0]) else: return template_name def patch_templates(): # type: () -> None from django.template.response import SimpleTemplateResponse from sentry_sdk.integrations.django import DjangoIntegration real_rendered_content = SimpleTemplateResponse.rendered_content @property # type: ignore def rendered_content(self): # type: (SimpleTemplateResponse) -> str hub = Hub.current if hub.get_integration(DjangoIntegration) is None: return real_rendered_content.fget(self) with hub.start_span( op=OP.TEMPLATE_RENDER, description=_get_template_name_description(self.template_name), ) as span: span.set_data("context", self.context_data) return real_rendered_content.fget(self) SimpleTemplateResponse.rendered_content = rendered_content if DJANGO_VERSION < (1, 7): return import django.shortcuts real_render = django.shortcuts.render @_functools.wraps(real_render) def render(request, template_name, context=None, args, kwargs): # type: (django.http.HttpRequest, str, Optional[Dict[str, Any]], Any, *Any) -> django.http.HttpResponse hub = Hub.current if hub.get_integration(DjangoIntegration) is None: return real_render(request, template_name, context, args, *kwargs) # Inject trace meta tags into template context context = context or {} if "sentry_trace_meta" not in context: context["sentry_trace_meta"] = mark_safe(hub.trace_propagation_meta()) with hub.start_span( op=OP.TEMPLATE_RENDER, description=_get_template_name_description(template_name), ) as span: span.set_data("context", context) return real_render(request, template_name, context, args, **kwargs) django.shortcuts.render = render def _get_template_frame_from_debug(debug): # type: (Dict[str, Any]) -> Dict[str, Any] if debug is None: return None lineno = debug["line"] filename = debug["name"] if filename is None: filename = "<django template>" pre_context = [] post_context = [] context_line = None for i, line in debug["source_lines"]: if i < lineno: pre_context.append(line) elif i > lineno: post_context.append(line) else: context_line = line return { "filename": filename, "lineno": lineno, "pre_context": pre_context[-5:], "post_context": post_context[:5], "context_line": context_line, "in_app": True, } def _linebreak_iter(template_source): # type: (str) -> Iterator[int] yield 0 p = template_source.find("\n") while p >= 0: yield p + 1 p = template_source.find("\n", p + 1) def _get_template_frame_from_source(source): # type: (Tuple[Origin, Tuple[int, int]]) -> Optional[Dict[str, Any]] if not source: return None origin, (start, end) = source filename = getattr(origin, "loadname", None) if filename is None: filename = "<django template>" template_source = origin.reload() lineno = None upto = 0 pre_context = [] post_context = [] context_line = None for num, next in enumerate(_linebreak_iter(template_source)): line = template_source[upto:next] if start >= upto and end <= next: lineno = num context_line = line elif lineno is None: pre_context.append(line) else: post_context.append(line) upto = next if context_line is None or lineno is None: return None return { "filename": filename, "lineno": lineno, "pre_context": pre_context[-5:], "post_context": post_context[:5], "context_line": context_line, }	5.56 KB	2026-01-20 12:58
📄 transactions.py """ Copied from raven-python. Used for `DjangoIntegration(transaction_fron="raven_legacy")`. """ from __future__ import absolute_import import re from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from django.urls.resolvers import URLResolver from typing import Dict from typing import List from typing import Optional from django.urls.resolvers import URLPattern from typing import Tuple from typing import Union from re import Pattern try: from django.urls import get_resolver except ImportError: from django.core.urlresolvers import get_resolver def get_regex(resolver_or_pattern): # type: (Union[URLPattern, URLResolver]) -> Pattern[str] """Utility method for django's deprecated resolver.regex""" try: regex = resolver_or_pattern.regex except AttributeError: regex = resolver_or_pattern.pattern.regex return regex class RavenResolver(object): _optional_group_matcher = re.compile(r"\(\?\:([^\)]+)\)") _named_group_matcher = re.compile(r"\(\?P<(\w+)>.*\)") _non_named_group_matcher = re.compile(r"\([^\)]+\)") # [foo\|bar\|baz] _either_option_matcher = re.compile(r"\[([^\]]+)\\|([^\]]+)\]") _camel_re = re.compile(r"([A-Z]+)([a-z])") _cache = {} # type: Dict[URLPattern, str] def _simplify(self, pattern): # type: (str) -> str r""" Clean up urlpattern regexes into something readable by humans: From: > "^(?P<sport_slug>\w+)/athletes/(?P<athlete_slug>\w+)/$" To: > "{sport_slug}/athletes/{athlete_slug}/" """ # remove optional params # TODO(dcramer): it'd be nice to change these into [%s] but it currently # conflicts with the other rules because we're doing regexp matches # rather than parsing tokens result = self._optional_group_matcher.sub(lambda m: "%s" % m.group(1), pattern) # handle named groups first result = self._named_group_matcher.sub(lambda m: "{%s}" % m.group(1), result) # handle non-named groups result = self._non_named_group_matcher.sub("{var}", result) # handle optional params result = self._either_option_matcher.sub(lambda m: m.group(1), result) # clean up any outstanding regex-y characters. result = ( result.replace("^", "") .replace("$", "") .replace("?", "") .replace("\\A", "") .replace("\\Z", "") .replace("//", "/") .replace("\\", "") ) return result def _resolve(self, resolver, path, parents=None): # type: (URLResolver, str, Optional[List[URLResolver]]) -> Optional[str] match = get_regex(resolver).search(path) # Django < 2.0 if not match: return None if parents is None: parents = [resolver] elif resolver not in parents: parents = parents + [resolver] new_path = path[match.end() :] for pattern in resolver.url_patterns: # this is an include() if not pattern.callback: match_ = self._resolve(pattern, new_path, parents) if match_: return match_ continue elif not get_regex(pattern).search(new_path): continue try: return self._cache[pattern] except KeyError: pass prefix = "".join(self._simplify(get_regex(p).pattern) for p in parents) result = prefix + self._simplify(get_regex(pattern).pattern) if not result.startswith("/"): result = "/" + result self._cache[pattern] = result return result return None def resolve( self, path, # type: str urlconf=None, # type: Union[None, Tuple[URLPattern, URLPattern, URLResolver], Tuple[URLPattern]] ): # type: (...) -> Optional[str] resolver = get_resolver(urlconf) match = self._resolve(resolver, path) return match LEGACY_RESOLVER = RavenResolver()	4.08 KB	2026-01-20 12:58
📄 views.py from sentry_sdk.consts import OP from sentry_sdk.hub import Hub from sentry_sdk._types import TYPE_CHECKING from sentry_sdk import _functools if TYPE_CHECKING: from typing import Any try: from asyncio import iscoroutinefunction except ImportError: iscoroutinefunction = None # type: ignore try: from sentry_sdk.integrations.django.asgi import wrap_async_view except (ImportError, SyntaxError): wrap_async_view = None # type: ignore def patch_views(): # type: () -> None from django.core.handlers.base import BaseHandler from django.template.response import SimpleTemplateResponse from sentry_sdk.integrations.django import DjangoIntegration old_make_view_atomic = BaseHandler.make_view_atomic old_render = SimpleTemplateResponse.render def sentry_patched_render(self): # type: (SimpleTemplateResponse) -> Any hub = Hub.current with hub.start_span( op=OP.VIEW_RESPONSE_RENDER, description="serialize response" ): return old_render(self) @_functools.wraps(old_make_view_atomic) def sentry_patched_make_view_atomic(self, args, kwargs): # type: (Any, Any, *Any) -> Any callback = old_make_view_atomic(self, args, *kwargs) # XXX: The wrapper function is created for every request. Find more # efficient way to wrap views (or build a cache?) hub = Hub.current integration = hub.get_integration(DjangoIntegration) if integration is not None and integration.middleware_spans: if ( iscoroutinefunction is not None and wrap_async_view is not None and iscoroutinefunction(callback) ): sentry_wrapped_callback = wrap_async_view(hub, callback) else: sentry_wrapped_callback = _wrap_sync_view(hub, callback) else: sentry_wrapped_callback = callback return sentry_wrapped_callback SimpleTemplateResponse.render = sentry_patched_render BaseHandler.make_view_atomic = sentry_patched_make_view_atomic def _wrap_sync_view(hub, callback): # type: (Hub, Any) -> Any @_functools.wraps(callback) def sentry_wrapped_callback(request, args, *kwargs): # type: (Any, Any, *Any) -> Any with hub.configure_scope() as sentry_scope: # set the active thread id to the handler thread for sync views # this isn't necessary for async views since that runs on main if sentry_scope.profile is not None: sentry_scope.profile.update_active_thread_id() with hub.start_span( op=OP.VIEW_RENDER, description=request.resolver_match.view_name ): return callback(request, args, **kwargs) return sentry_wrapped_callback	2.8 KB	2026-01-20 12:58

Terminal