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

from __future__ import absolute_import

from sentry_sdk._compat import text_type
from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk.consts import SPANDATA
from sentry_sdk.hub import Hub
from sentry_sdk.integrations import Integration, DidNotEnable
from sentry_sdk.tracing_utils import record_sql_queries

from sentry_sdk.utils import parse_version

try:
    from sqlalchemy.engine import Engine  # type: ignore
    from sqlalchemy.event import listen  # type: ignore
    from sqlalchemy import __version__ as SQLALCHEMY_VERSION  # type: ignore
except ImportError:
    raise DidNotEnable("SQLAlchemy not installed.")

if TYPE_CHECKING:
    from typing import Any
    from typing import ContextManager
    from typing import Optional

from sentry_sdk.tracing import Span

class SqlalchemyIntegration(Integration):
    identifier = "sqlalchemy"

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

version = parse_version(SQLALCHEMY_VERSION)

if version is None:
            raise DidNotEnable(
                "Unparsable SQLAlchemy version: {}".format(SQLALCHEMY_VERSION)
            )

if version < (1, 2):
            raise DidNotEnable("SQLAlchemy 1.2 or newer required.")

listen(Engine, "before_cursor_execute", _before_cursor_execute)
        listen(Engine, "after_cursor_execute", _after_cursor_execute)
        listen(Engine, "handle_error", _handle_error)

def _before_cursor_execute(
    conn, cursor, statement, parameters, context, executemany, *args
):
    # type: (Any, Any, Any, Any, Any, bool, *Any) -> None
    hub = Hub.current
    if hub.get_integration(SqlalchemyIntegration) is None:
        return

ctx_mgr = record_sql_queries(
        hub,
        cursor,
        statement,
        parameters,
        paramstyle=context and context.dialect and context.dialect.paramstyle or None,
        executemany=executemany,
    )
    context._sentry_sql_span_manager = ctx_mgr

span = ctx_mgr.__enter__()

if span is not None:
        _set_db_data(span, conn)
        context._sentry_sql_span = span

def _after_cursor_execute(conn, cursor, statement, parameters, context, *args):
    # type: (Any, Any, Any, Any, Any, *Any) -> None
    ctx_mgr = getattr(
        context, "_sentry_sql_span_manager", None
    )  # type: Optional[ContextManager[Any]]

if ctx_mgr is not None:
        context._sentry_sql_span_manager = None
        ctx_mgr.__exit__(None, None, None)

def _handle_error(context, *args):
    # type: (Any, *Any) -> None
    execution_context = context.execution_context
    if execution_context is None:
        return

span = getattr(execution_context, "_sentry_sql_span", None)  # type: Optional[Span]

if span is not None:
        span.set_status("internal_error")

# _after_cursor_execute does not get called for crashing SQL stmts. Judging
    # from SQLAlchemy codebase it does seem like any error coming into this
    # handler is going to be fatal.
    ctx_mgr = getattr(
        execution_context, "_sentry_sql_span_manager", None
    )  # type: Optional[ContextManager[Any]]

if ctx_mgr is not None:
        execution_context._sentry_sql_span_manager = None
        ctx_mgr.__exit__(None, None, None)

# See: https://docs.sqlalchemy.org/en/20/dialects/index.html
def _get_db_system(name):
    # type: (str) -> Optional[str]
    name = text_type(name)

if "sqlite" in name:
        return "sqlite"

if "postgres" in name:
        return "postgresql"

if "mariadb" in name:
        return "mariadb"

if "mysql" in name:
        return "mysql"

if "oracle" in name:
        return "oracle"

return None

def _set_db_data(span, conn):
    # type: (Span, Any) -> None
    db_system = _get_db_system(conn.engine.name)
    if db_system is not None:
        span.set_data(SPANDATA.DB_SYSTEM, db_system)

db_name = conn.engine.url.database
    if db_name is not None:
        span.set_data(SPANDATA.DB_NAME, db_name)

server_address = conn.engine.url.host
    if server_address is not None:
        span.set_data(SPANDATA.SERVER_ADDRESS, server_address)

server_port = conn.engine.url.port
    if server_port is not None:
        span.set_data(SPANDATA.SERVER_PORT, server_port)

X7ROOT File Manager

Editing: sqlalchemy.py

Upload File

Create Folder