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: asyncio.py

from __future__ import absolute_import
import sys

from sentry_sdk._compat import reraise
from sentry_sdk.consts import OP
from sentry_sdk.hub import Hub
from sentry_sdk.integrations import Integration, DidNotEnable
from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk.utils import event_from_exception

try:
    import asyncio
    from asyncio.tasks import Task
except ImportError:
    raise DidNotEnable("asyncio not available")

if TYPE_CHECKING:
    from typing import Any

from sentry_sdk._types import ExcInfo

def get_name(coro):
    # type: (Any) -> str
    return (
        getattr(coro, "__qualname__", None)
        or getattr(coro, "__name__", None)
        or "coroutine without __name__"
    )

def patch_asyncio():
    # type: () -> None
    orig_task_factory = None
    try:
        loop = asyncio.get_running_loop()
        orig_task_factory = loop.get_task_factory()

def _sentry_task_factory(loop, coro):
            # type: (Any, Any) -> Any

async def _coro_creating_hub_and_span():
                # type: () -> Any
                hub = Hub(Hub.current)
                result = None

with hub:
                    with hub.start_span(op=OP.FUNCTION, description=get_name(coro)):
                        try:
                            result = await coro
                        except Exception:
                            reraise(*_capture_exception(hub))

return result

# Trying to use user set task factory (if there is one)
            if orig_task_factory:
                return orig_task_factory(loop, _coro_creating_hub_and_span())

# The default task factory in `asyncio` does not have its own function
            # but is just a couple of lines in `asyncio.base_events.create_task()`
            # Those lines are copied here.

# WARNING:
            # If the default behavior of the task creation in asyncio changes,
            # this will break!
            task = Task(_coro_creating_hub_and_span(), loop=loop)
            if task._source_traceback:  # type: ignore
                del task._source_traceback[-1]  # type: ignore

return task

loop.set_task_factory(_sentry_task_factory)
    except RuntimeError:
        # When there is no running loop, we have nothing to patch.
        pass

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

integration = hub.get_integration(AsyncioIntegration)
    if integration 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": "asyncio", "handled": False},
        )
        hub.capture_event(event, hint=hint)

return exc_info

class AsyncioIntegration(Integration):
    identifier = "asyncio"

@staticmethod
    def setup_once():
        # type: () -> None
        patch_asyncio()

X7ROOT File Manager

Editing: asyncio.py

Upload File

Create Folder