File Manager

X7ROOT File Manager

Current Path: /opt/hc_python/lib/python3.12/site-packages/sentry_sdk/integrations

opt / hc_python / lib / python3.12 / site-packages / sentry_sdk / integrations /

📁 ..
📄 __init__.py(9.98 KB)
📁 __pycache__
📄 _asgi_common.py(3.11 KB)
📄 _wsgi_common.py(7.38 KB)
📄 aiohttp.py(12.59 KB)
📄 anthropic.py(9.21 KB)
📄 argv.py(911 B)
📄 ariadne.py(5.7 KB)
📄 arq.py(7.67 KB)
📄 asgi.py(12.48 KB)
📄 asyncio.py(3.94 KB)
📄 asyncpg.py(6.37 KB)
📄 atexit.py(1.61 KB)
📄 aws_lambda.py(17.53 KB)
📄 beam.py(5.06 KB)
📄 boto3.py(4.31 KB)
📄 bottle.py(6.46 KB)
📁 celery
📄 chalice.py(4.59 KB)
📄 clickhouse_driver.py(5.12 KB)
📄 cloud_resource_context.py(7.6 KB)
📄 cohere.py(9.11 KB)
📄 dedupe.py(1.38 KB)
📁 django
📄 dramatiq.py(5.45 KB)
📄 excepthook.py(2.35 KB)
📄 executing.py(1.95 KB)
📄 falcon.py(9.28 KB)
📄 fastapi.py(4.61 KB)
📄 flask.py(8.54 KB)
📄 gcp.py(8.08 KB)
📄 gnu_backtrace.py(2.83 KB)
📄 gql.py(4.08 KB)
📄 graphene.py(4.92 KB)
📁 grpc
📄 httpx.py(5.73 KB)
📄 huey.py(5.32 KB)
📄 huggingface_hub.py(6.4 KB)
📄 langchain.py(17.3 KB)
📄 launchdarkly.py(1.89 KB)
📄 litestar.py(11.3 KB)
📄 logging.py(13.19 KB)
📄 loguru.py(3.54 KB)
📄 modules.py(820 B)
📄 openai.py(15.22 KB)
📄 openfeature.py(1.21 KB)
📁 opentelemetry
📄 pure_eval.py(4.47 KB)
📄 pymongo.py(6.23 KB)
📄 pyramid.py(7.19 KB)
📄 quart.py(7.26 KB)
📄 ray.py(4.06 KB)
📁 redis
📄 rq.py(5.18 KB)
📄 rust_tracing.py(8.87 KB)
📄 sanic.py(12.66 KB)
📄 serverless.py(1.76 KB)
📄 socket.py(3.09 KB)
📁 spark
📄 sqlalchemy.py(4.27 KB)
📄 starlette.py(25.79 KB)
📄 starlite.py(10.37 KB)
📄 statsig.py(1.2 KB)
📄 stdlib.py(8.62 KB)
📄 strawberry.py(13.79 KB)
📄 sys_exit.py(2.43 KB)
📄 threading.py(5.27 KB)
📄 tornado.py(7.05 KB)
📄 trytond.py(1.61 KB)
📄 typer.py(1.77 KB)
📄 unleash.py(1.03 KB)
📄 wsgi.py(10.5 KB)

Editing: clickhouse_driver.py

import sentry_sdk
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations import _check_minimum_version, Integration, DidNotEnable
from sentry_sdk.tracing import Span
from sentry_sdk.scope import should_send_default_pii
from sentry_sdk.utils import capture_internal_exceptions, ensure_integration_enabled

from typing import TYPE_CHECKING, TypeVar

# Hack to get new Python features working in older versions
# without introducing a hard dependency on `typing_extensions`
# from: https://stackoverflow.com/a/71944042/300572
if TYPE_CHECKING:
    from typing import ParamSpec, Callable
else:
    # Fake ParamSpec
    class ParamSpec:
        def __init__(self, _):
            self.args = None
            self.kwargs = None

# Callable[anything] will return None
    class _Callable:
        def __getitem__(self, _):
            return None

# Make instances
    Callable = _Callable()

try:
    import clickhouse_driver  # type: ignore[import-not-found]

except ImportError:
    raise DidNotEnable("clickhouse-driver not installed.")

class ClickhouseDriverIntegration(Integration):
    identifier = "clickhouse_driver"
    origin = f"auto.db.{identifier}"

@staticmethod
    def setup_once() -> None:
        _check_minimum_version(ClickhouseDriverIntegration, clickhouse_driver.VERSION)

# Every query is done using the Connection's `send_query` function
        clickhouse_driver.connection.Connection.send_query = _wrap_start(
            clickhouse_driver.connection.Connection.send_query
        )

# If the query contains parameters then the send_data function is used to send those parameters to clickhouse
        clickhouse_driver.client.Client.send_data = _wrap_send_data(
            clickhouse_driver.client.Client.send_data
        )

# Every query ends either with the Client's `receive_end_of_query` (no result expected)
        # or its `receive_result` (result expected)
        clickhouse_driver.client.Client.receive_end_of_query = _wrap_end(
            clickhouse_driver.client.Client.receive_end_of_query
        )
        if hasattr(clickhouse_driver.client.Client, "receive_end_of_insert_query"):
            # In 0.2.7, insert queries are handled separately via `receive_end_of_insert_query`
            clickhouse_driver.client.Client.receive_end_of_insert_query = _wrap_end(
                clickhouse_driver.client.Client.receive_end_of_insert_query
            )
        clickhouse_driver.client.Client.receive_result = _wrap_end(
            clickhouse_driver.client.Client.receive_result
        )

P = ParamSpec("P")
T = TypeVar("T")

def _wrap_start(f: Callable[P, T]) -> Callable[P, T]:
    @ensure_integration_enabled(ClickhouseDriverIntegration, f)
    def _inner(*args: P.args, **kwargs: P.kwargs) -> T:
        connection = args[0]
        query = args[1]
        query_id = args[2] if len(args) > 2 else kwargs.get("query_id")
        params = args[3] if len(args) > 3 else kwargs.get("params")

span = sentry_sdk.start_span(
            op=OP.DB,
            name=query,
            origin=ClickhouseDriverIntegration.origin,
        )

connection._sentry_span = span  # type: ignore[attr-defined]

_set_db_data(span, connection)

span.set_data("query", query)

if query_id:
            span.set_data("db.query_id", query_id)

if params and should_send_default_pii():
            span.set_data("db.params", params)

# run the original code
        ret = f(*args, **kwargs)

return ret

return _inner

def _wrap_end(f: Callable[P, T]) -> Callable[P, T]:
    def _inner_end(*args: P.args, **kwargs: P.kwargs) -> T:
        res = f(*args, **kwargs)
        instance = args[0]
        span = getattr(instance.connection, "_sentry_span", None)  # type: ignore[attr-defined]

if span is not None:
            if res is not None and should_send_default_pii():
                span.set_data("db.result", res)

with capture_internal_exceptions():
                span.scope.add_breadcrumb(
                    message=span._data.pop("query"), category="query", data=span._data
                )

span.finish()

return res

return _inner_end

def _wrap_send_data(f: Callable[P, T]) -> Callable[P, T]:
    def _inner_send_data(*args: P.args, **kwargs: P.kwargs) -> T:
        instance = args[0]  # type: clickhouse_driver.client.Client
        data = args[2]
        span = getattr(instance.connection, "_sentry_span", None)

if span is not None:
            _set_db_data(span, instance.connection)

if should_send_default_pii():
                db_params = span._data.get("db.params", [])
                db_params.extend(data)
                span.set_data("db.params", db_params)

return f(*args, **kwargs)

return _inner_send_data

def _set_db_data(
    span: Span, connection: clickhouse_driver.connection.Connection
) -> None:
    span.set_data(SPANDATA.DB_SYSTEM, "clickhouse")
    span.set_data(SPANDATA.SERVER_ADDRESS, connection.host)
    span.set_data(SPANDATA.SERVER_PORT, connection.port)
    span.set_data(SPANDATA.DB_NAME, connection.database)
    span.set_data(SPANDATA.DB_USER, connection.user)

X7ROOT File Manager

Editing: clickhouse_driver.py

Upload File

Create Folder