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

from __future__ import absolute_import

import socket
from sentry_sdk import Hub
from sentry_sdk._types import MYPY
from sentry_sdk.consts import OP
from sentry_sdk.integrations import Integration

if MYPY:
    from socket import AddressFamily, SocketKind
    from typing import Tuple, Optional, Union, List

__all__ = ["SocketIntegration"]

class SocketIntegration(Integration):
    identifier = "socket"

@staticmethod
    def setup_once():
        # type: () -> None
        """
        patches two of the most used functions of socket: create_connection and getaddrinfo(dns resolver)
        """
        _patch_create_connection()
        _patch_getaddrinfo()

def _get_span_description(host, port):
    # type: (Union[bytes, str, None], Union[str, int, None]) -> str

try:
        host = host.decode()  # type: ignore
    except (UnicodeDecodeError, AttributeError):
        pass

description = "%s:%s" % (host, port)  # type: ignore

return description

def _patch_create_connection():
    # type: () -> None
    real_create_connection = socket.create_connection

def create_connection(
        address,
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT,  # type: ignore
        source_address=None,
    ):
        # type: (Tuple[Optional[str], int], Optional[float], Optional[Tuple[Union[bytearray, bytes, str], int]])-> socket.socket
        hub = Hub.current
        if hub.get_integration(SocketIntegration) is None:
            return real_create_connection(
                address=address, timeout=timeout, source_address=source_address
            )

with hub.start_span(
            op=OP.SOCKET_CONNECTION,
            description=_get_span_description(address[0], address[1]),
        ) as span:
            span.set_data("address", address)
            span.set_data("timeout", timeout)
            span.set_data("source_address", source_address)

return real_create_connection(
                address=address, timeout=timeout, source_address=source_address
            )

socket.create_connection = create_connection  # type: ignore

def _patch_getaddrinfo():
    # type: () -> None
    real_getaddrinfo = socket.getaddrinfo

def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
        # type: (Union[bytes, str, None], Union[str, int, None], int, int, int, int) -> List[Tuple[AddressFamily, SocketKind, int, str, Union[Tuple[str, int], Tuple[str, int, int, int]]]]
        hub = Hub.current
        if hub.get_integration(SocketIntegration) is None:
            return real_getaddrinfo(host, port, family, type, proto, flags)

with hub.start_span(
            op=OP.SOCKET_DNS, description=_get_span_description(host, port)
        ) as span:
            span.set_data("host", host)
            span.set_data("port", port)

return real_getaddrinfo(host, port, family, type, proto, flags)

socket.getaddrinfo = getaddrinfo  # type: ignore

X7ROOT File Manager

Editing: socket.py

Upload File

Create Folder