File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/sentry_sdk/integrations

opt / cloudlinux / venv / lib / python3.11 / site-packages / sentry_sdk / integrations /

?? ..
?? __init__.py(6.67 KB)
?? __pycache__
?? _wsgi_common.py(4.43 KB)
?? aiohttp.py(11.28 KB)
?? argv.py(963 B)
?? arq.py(6.58 KB)
?? asgi.py(11.54 KB)
?? asyncio.py(2.98 KB)
?? atexit.py(1.8 KB)
?? aws_lambda.py(15.44 KB)
?? beam.py(5.56 KB)
?? boto3.py(4.44 KB)
?? bottle.py(6.32 KB)
?? celery.py(18.65 KB)
?? chalice.py(4.66 KB)
?? cloud_resource_context.py(6.6 KB)
?? dedupe.py(1.16 KB)
?? django
?? excepthook.py(2.21 KB)
?? executing.py(1.99 KB)
?? falcon.py(7.8 KB)
?? fastapi.py(4.39 KB)
?? flask.py(7.72 KB)
?? gcp.py(8.02 KB)
?? gnu_backtrace.py(2.86 KB)
?? grpc
?? httpx.py(4.89 KB)
?? huey.py(4.59 KB)
?? logging.py(8.97 KB)
?? loguru.py(2.98 KB)
?? modules.py(2.06 KB)
?? opentelemetry
?? pure_eval.py(4.45 KB)
?? pymongo.py(5.87 KB)
?? pyramid.py(7.27 KB)
?? quart.py(7.2 KB)
?? redis
?? rq.py(5.28 KB)
?? sanic.py(11.06 KB)
?? serverless.py(1.93 KB)
?? socket.py(2.88 KB)
?? spark
?? sqlalchemy.py(4.14 KB)
?? starlette.py(22.67 KB)
?? starlite.py(9.85 KB)
?? stdlib.py(8.06 KB)
?? threading.py(2.87 KB)
?? tornado.py(7.17 KB)
?? trytond.py(1.7 KB)
?? wsgi.py(9.36 KB)

Editing: threading.py

from __future__ import absolute_import

import sys
from functools import wraps
from threading import Thread, current_thread

from sentry_sdk import Hub
from sentry_sdk._compat import reraise
from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk.integrations import Integration
from sentry_sdk.utils import event_from_exception, capture_internal_exceptions

if TYPE_CHECKING:
    from typing import Any
    from typing import TypeVar
    from typing import Callable
    from typing import Optional

from sentry_sdk._types import ExcInfo

F = TypeVar("F", bound=Callable[..., Any])

class ThreadingIntegration(Integration):
    identifier = "threading"

def __init__(self, propagate_hub=False):
        # type: (bool) -> None
        self.propagate_hub = propagate_hub

@staticmethod
    def setup_once():
        # type: () -> None
        old_start = Thread.start

@wraps(old_start)
        def sentry_start(self, *a, **kw):
            # type: (Thread, *Any, **Any) -> Any
            hub = Hub.current
            integration = hub.get_integration(ThreadingIntegration)
            if integration is not None:
                if not integration.propagate_hub:
                    hub_ = None
                else:
                    hub_ = Hub(hub)
                # Patching instance methods in `start()` creates a reference cycle if
                # done in a naive way. See
                # https://github.com/getsentry/sentry-python/pull/434
                #
                # In threading module, using current_thread API will access current thread instance
                # without holding it to avoid a reference cycle in an easier way.
                with capture_internal_exceptions():
                    new_run = _wrap_run(hub_, getattr(self.run, "__func__", self.run))
                    self.run = new_run  # type: ignore

return old_start(self, *a, **kw)

Thread.start = sentry_start  # type: ignore

def _wrap_run(parent_hub, old_run_func):
    # type: (Optional[Hub], F) -> F
    @wraps(old_run_func)
    def run(*a, **kw):
        # type: (*Any, **Any) -> Any
        hub = parent_hub or Hub.current
        with hub:
            try:
                self = current_thread()
                return old_run_func(self, *a, **kw)
            except Exception:
                reraise(*_capture_exception())

return run  # type: ignore

def _capture_exception():
    # type: () -> ExcInfo
    hub = Hub.current
    exc_info = sys.exc_info()

if hub.get_integration(ThreadingIntegration) is not None:
        # If an integration is there, a client has to be there.
        client = hub.client  # type: Any

event, hint = event_from_exception(
            exc_info,
            client_options=client.options,
            mechanism={"type": "threading", "handled": False},
        )
        hub.capture_event(event, hint=hint)

return exc_info

X7ROOT File Manager

Editing: threading.py

Upload File

Create Folder