path: root/.venv/lib/python3.12/site-packages/opentelemetry/instrumentation/dbapi/__init__.py

# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
The trace integration with Database API supports libraries that follow the
Python Database API Specification v2.0.
`<https://www.python.org/dev/peps/pep-0249/>`_

Usage
-----

.. code-block:: python

    import mysql.connector
    import pyodbc

    from opentelemetry.instrumentation.dbapi import trace_integration


    # Ex: mysql.connector
    trace_integration(mysql.connector, "connect", "mysql")
    # Ex: pyodbc
    trace_integration(pyodbc, "Connection", "odbc")

API
---
"""

from __future__ import annotations

import functools
import logging
import re
from typing import Any, Callable, Generic, TypeVar

import wrapt
from wrapt import wrap_function_wrapper

from opentelemetry import trace as trace_api
from opentelemetry.instrumentation.dbapi.version import __version__
from opentelemetry.instrumentation.sqlcommenter_utils import _add_sql_comment
from opentelemetry.instrumentation.utils import (
    _get_opentelemetry_values,
    unwrap,
)
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.trace import SpanKind, TracerProvider, get_tracer
from opentelemetry.util._importlib_metadata import version as util_version

_DB_DRIVER_ALIASES = {
    "MySQLdb": "mysqlclient",
}

_logger = logging.getLogger(__name__)

ConnectionT = TypeVar("ConnectionT")
CursorT = TypeVar("CursorT")


def trace_integration(
    connect_module: Callable[..., Any],
    connect_method_name: str,
    database_system: str,
    connection_attributes: dict[str, Any] | None = None,
    tracer_provider: TracerProvider | None = None,
    capture_parameters: bool = False,
    enable_commenter: bool = False,
    db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
    enable_attribute_commenter: bool = False,
):
    """Integrate with DB API library.
    https://www.python.org/dev/peps/pep-0249/

    Args:
        connect_module: Module name where connect method is available.
        connect_method_name: The connect method name.
        database_system: An identifier for the database management system (DBMS)
            product being used.
        connection_attributes: Attribute names for database, port, host and
            user in Connection object.
        tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
            use. If omitted the current configured one is used.
        capture_parameters: Configure if db.statement.parameters should be captured.
        enable_commenter: Flag to enable/disable sqlcommenter.
        db_api_integration_factory: The `DatabaseApiIntegration` to use. If none is passed the
            default one is used.
        enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` span attribute. Only available if enable_commenter=True.
    """
    wrap_connect(
        __name__,
        connect_module,
        connect_method_name,
        database_system,
        connection_attributes,
        version=__version__,
        tracer_provider=tracer_provider,
        capture_parameters=capture_parameters,
        enable_commenter=enable_commenter,
        db_api_integration_factory=db_api_integration_factory,
        enable_attribute_commenter=enable_attribute_commenter,
    )


def wrap_connect(
    name: str,
    connect_module: Callable[..., Any],
    connect_method_name: str,
    database_system: str,
    connection_attributes: dict[str, Any] | None = None,
    version: str = "",
    tracer_provider: TracerProvider | None = None,
    capture_parameters: bool = False,
    enable_commenter: bool = False,
    db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
    commenter_options: dict[str, Any] | None = None,
    enable_attribute_commenter: bool = False,
):
    """Integrate with DB API library.
    https://www.python.org/dev/peps/pep-0249/

    Args:
        connect_module: Module name where connect method is available.
        connect_method_name: The connect method name.
        database_system: An identifier for the database management system (DBMS)
            product being used.
        connection_attributes: Attribute names for database, port, host and
            user in Connection object.
        tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
            use. If omitted the current configured one is used.
        capture_parameters: Configure if db.statement.parameters should be captured.
        enable_commenter: Flag to enable/disable sqlcommenter.
        db_api_integration_factory: The `DatabaseApiIntegration` to use. If none is passed the
            default one is used.
        commenter_options: Configurations for tags to be appended at the sql query.
        enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` span attribute. Only available if enable_commenter=True.

    """
    db_api_integration_factory = (
        db_api_integration_factory or DatabaseApiIntegration
    )

    # pylint: disable=unused-argument
    def wrap_connect_(
        wrapped: Callable[..., Any],
        instance: Any,
        args: tuple[Any, Any],
        kwargs: dict[Any, Any],
    ):
        db_integration = db_api_integration_factory(
            name,
            database_system,
            connection_attributes=connection_attributes,
            version=version,
            tracer_provider=tracer_provider,
            capture_parameters=capture_parameters,
            enable_commenter=enable_commenter,
            commenter_options=commenter_options,
            connect_module=connect_module,
            enable_attribute_commenter=enable_attribute_commenter,
        )
        return db_integration.wrapped_connection(wrapped, args, kwargs)

    try:
        wrap_function_wrapper(
            connect_module, connect_method_name, wrap_connect_
        )
    except Exception as ex:  # pylint: disable=broad-except
        _logger.warning("Failed to integrate with DB API. %s", str(ex))


def unwrap_connect(
    connect_module: Callable[..., Any], connect_method_name: str
):
    """Disable integration with DB API library.
    https://www.python.org/dev/peps/pep-0249/

    Args:
        connect_module: Module name where the connect method is available.
        connect_method_name: The connect method name.
    """
    unwrap(connect_module, connect_method_name)


def instrument_connection(
    name: str,
    connection: ConnectionT | TracedConnectionProxy[ConnectionT],
    database_system: str,
    connection_attributes: dict[str, Any] | None = None,
    version: str = "",
    tracer_provider: TracerProvider | None = None,
    capture_parameters: bool = False,
    enable_commenter: bool = False,
    commenter_options: dict[str, Any] | None = None,
    connect_module: Callable[..., Any] | None = None,
    enable_attribute_commenter: bool = False,
    db_api_integration_factory: type[DatabaseApiIntegration] | None = None,
) -> TracedConnectionProxy[ConnectionT]:
    """Enable instrumentation in a database connection.

    Args:
        name: The instrumentation module name.
        connection: The connection to instrument.
        database_system: An identifier for the database management system (DBMS)
            product being used.
        connection_attributes: Attribute names for database, port, host and
            user in a connection object.
        tracer_provider: The :class:`opentelemetry.trace.TracerProvider` to
            use. If omitted the current configured one is used.
        capture_parameters: Configure if db.statement.parameters should be captured.
        enable_commenter: Flag to enable/disable sqlcommenter.
        commenter_options: Configurations for tags to be appended at the sql query.
        connect_module: Module name where connect method is available.
        enable_attribute_commenter: Flag to enable/disable sqlcomment inclusion in `db.statement` span attribute. Only available if enable_commenter=True.
        db_api_integration_factory: A class or factory function to use as a
            replacement for :class:`DatabaseApiIntegration`. Can be used to
            obtain connection attributes from the connect method instead of
            from the connection itself (as done by the pymssql intrumentor).

    Returns:
        An instrumented connection.
    """
    if isinstance(connection, wrapt.ObjectProxy):
        _logger.warning("Connection already instrumented")
        return connection

    db_api_integration_factory = (
        db_api_integration_factory or DatabaseApiIntegration
    )

    db_integration = db_api_integration_factory(
        name,
        database_system,
        connection_attributes=connection_attributes,
        version=version,
        tracer_provider=tracer_provider,
        capture_parameters=capture_parameters,
        enable_commenter=enable_commenter,
        commenter_options=commenter_options,
        connect_module=connect_module,
        enable_attribute_commenter=enable_attribute_commenter,
    )
    db_integration.get_connection_attributes(connection)
    return get_traced_connection_proxy(connection, db_integration)


def uninstrument_connection(
    connection: ConnectionT | TracedConnectionProxy[ConnectionT],
) -> ConnectionT:
    """Disable instrumentation in a database connection.

    Args:
        connection: The connection to uninstrument.

    Returns:
        An uninstrumented connection.
    """
    if isinstance(connection, wrapt.ObjectProxy):
        return connection.__wrapped__

    _logger.warning("Connection is not instrumented")
    return connection


class DatabaseApiIntegration:
    def __init__(
        self,
        name: str,
        database_system: str,
        connection_attributes: dict[str, Any] | None = None,
        version: str = "",
        tracer_provider: TracerProvider | None = None,
        capture_parameters: bool = False,
        enable_commenter: bool = False,
        commenter_options: dict[str, Any] | None = None,
        connect_module: Callable[..., Any] | None = None,
        enable_attribute_commenter: bool = False,
    ):
        if connection_attributes is None:
            self.connection_attributes = {
                "database": "database",
                "port": "port",
                "host": "host",
                "user": "user",
            }
        else:
            self.connection_attributes = connection_attributes
        self._name = name
        self._version = version
        self._tracer = get_tracer(
            self._name,
            instrumenting_library_version=self._version,
            tracer_provider=tracer_provider,
            schema_url="https://opentelemetry.io/schemas/1.11.0",
        )
        self.capture_parameters = capture_parameters
        self.enable_commenter = enable_commenter
        self.commenter_options = commenter_options
        self.enable_attribute_commenter = enable_attribute_commenter
        self.database_system = database_system
        self.connection_props: dict[str, Any] = {}
        self.span_attributes: dict[str, Any] = {}
        self.name = ""
        self.database = ""
        self.connect_module = connect_module
        self.commenter_data = self.calculate_commenter_data()

    def _get_db_version(self, db_driver: str) -> str:
        if db_driver in _DB_DRIVER_ALIASES:
            return util_version(_DB_DRIVER_ALIASES[db_driver])
        db_version = ""
        try:
            db_version = self.connect_module.__version__
        except AttributeError:
            db_version = "unknown"
        return db_version

    def calculate_commenter_data(self) -> dict[str, Any]:
        commenter_data: dict[str, Any] = {}
        if not self.enable_commenter:
            return commenter_data

        db_driver = getattr(self.connect_module, "__name__", "unknown")
        db_version = self._get_db_version(db_driver)

        commenter_data = {
            "db_driver": f"{db_driver}:{db_version.split(' ')[0]}",
            # PEP 249-compliant drivers should have the following attributes.
            # We can assume apilevel "1.0" if not given.
            # We use "unknown" for others to prevent uncaught AttributeError.
            # https://peps.python.org/pep-0249/#globals
            "dbapi_threadsafety": getattr(
                self.connect_module, "threadsafety", "unknown"
            ),
            "dbapi_level": getattr(self.connect_module, "apilevel", "1.0"),
            "driver_paramstyle": getattr(
                self.connect_module, "paramstyle", "unknown"
            ),
        }

        if self.database_system == "postgresql":
            if hasattr(self.connect_module, "__libpq_version__"):
                libpq_version = self.connect_module.__libpq_version__
            else:
                libpq_version = self.connect_module.pq.__build_version__
            commenter_data.update({"libpq_version": libpq_version})
        elif self.database_system == "mysql":
            mysqlc_version = ""
            if db_driver == "MySQLdb":
                mysqlc_version = self.connect_module._mysql.get_client_info()
            elif db_driver == "pymysql":
                mysqlc_version = self.connect_module.get_client_info()

            commenter_data.update({"mysql_client_version": mysqlc_version})

        return commenter_data

    def wrapped_connection(
        self,
        connect_method: Callable[..., ConnectionT],
        args: tuple[Any, ...],
        kwargs: dict[Any, Any],
    ) -> TracedConnectionProxy[ConnectionT]:
        """Add object proxy to connection object."""
        connection = connect_method(*args, **kwargs)
        self.get_connection_attributes(connection)
        return get_traced_connection_proxy(connection, self)

    def get_connection_attributes(self, connection: object) -> None:
        # Populate span fields using connection
        for key, value in self.connection_attributes.items():
            # Allow attributes nested in connection object
            attribute = functools.reduce(
                lambda attribute, attribute_value: getattr(
                    attribute, attribute_value, None
                ),
                value.split("."),
                connection,
            )
            if attribute:
                self.connection_props[key] = attribute
        self.name = self.database_system
        self.database = self.connection_props.get("database", "")
        if self.database:
            # PyMySQL encodes names with utf-8
            if hasattr(self.database, "decode"):
                self.database = self.database.decode(errors="ignore")
            self.name += "." + self.database
        user = self.connection_props.get("user")
        # PyMySQL encodes this data
        if user and isinstance(user, bytes):
            user = user.decode()
        if user is not None:
            self.span_attributes[SpanAttributes.DB_USER] = str(user)
        host = self.connection_props.get("host")
        if host is not None:
            self.span_attributes[SpanAttributes.NET_PEER_NAME] = host
        port = self.connection_props.get("port")
        if port is not None:
            self.span_attributes[SpanAttributes.NET_PEER_PORT] = port


# pylint: disable=abstract-method
class TracedConnectionProxy(wrapt.ObjectProxy, Generic[ConnectionT]):
    # pylint: disable=unused-argument
    def __init__(
        self,
        connection: ConnectionT,
        db_api_integration: DatabaseApiIntegration | None = None,
    ):
        wrapt.ObjectProxy.__init__(self, connection)
        self._self_db_api_integration = db_api_integration

    def __getattribute__(self, name: str):
        if object.__getattribute__(self, name):
            return object.__getattribute__(self, name)

        return object.__getattribute__(
            object.__getattribute__(self, "_connection"), name
        )

    def cursor(self, *args: Any, **kwargs: Any):
        return get_traced_cursor_proxy(
            self.__wrapped__.cursor(*args, **kwargs),
            self._self_db_api_integration,
        )

    def __enter__(self):
        self.__wrapped__.__enter__()
        return self

    def __exit__(self, *args: Any, **kwargs: Any):
        self.__wrapped__.__exit__(*args, **kwargs)


def get_traced_connection_proxy(
    connection: ConnectionT,
    db_api_integration: DatabaseApiIntegration | None,
    *args: Any,
    **kwargs: Any,
) -> TracedConnectionProxy[ConnectionT]:
    return TracedConnectionProxy(connection, db_api_integration)


class CursorTracer(Generic[CursorT]):
    def __init__(self, db_api_integration: DatabaseApiIntegration) -> None:
        self._db_api_integration = db_api_integration
        self._commenter_enabled = self._db_api_integration.enable_commenter
        self._commenter_options = (
            self._db_api_integration.commenter_options
            if self._db_api_integration.commenter_options
            else {}
        )
        self._enable_attribute_commenter = (
            self._db_api_integration.enable_attribute_commenter
        )
        self._connect_module = self._db_api_integration.connect_module
        self._leading_comment_remover = re.compile(r"^/\*.*?\*/")

    def _capture_mysql_version(self, cursor) -> None:
        """Lazy capture of mysql-connector client version using cursor, if applicable"""
        if (
            self._db_api_integration.database_system == "mysql"
            and self._db_api_integration.connect_module.__name__
            == "mysql.connector"
            and not self._db_api_integration.commenter_data[
                "mysql_client_version"
            ]
        ):
            self._db_api_integration.commenter_data["mysql_client_version"] = (
                cursor._cnx._cmysql.get_client_info()
            )

    def _get_commenter_data(self) -> dict:
        """Uses DB-API integration to return commenter data for sqlcomment"""
        commenter_data = dict(self._db_api_integration.commenter_data)
        if self._commenter_options.get("opentelemetry_values", True):
            commenter_data.update(**_get_opentelemetry_values())
        return {
            k: v
            for k, v in commenter_data.items()
            if self._commenter_options.get(k, True)
        }

    def _update_args_with_added_sql_comment(self, args, cursor) -> tuple:
        """Updates args with cursor info and adds sqlcomment to query statement"""
        try:
            args_list = list(args)
            self._capture_mysql_version(cursor)
            commenter_data = self._get_commenter_data()
            statement = _add_sql_comment(args_list[0], **commenter_data)
            args_list[0] = statement
            args = tuple(args_list)
        except Exception as exc:  # pylint: disable=broad-except
            _logger.exception(
                "Exception while generating sql comment: %s", exc
            )
        return args

    def _populate_span(
        self,
        span: trace_api.Span,
        cursor: CursorT,
        *args: tuple[Any, ...],
    ):
        if not span.is_recording():
            return
        statement = self.get_statement(cursor, args)
        span.set_attribute(
            SpanAttributes.DB_SYSTEM, self._db_api_integration.database_system
        )
        span.set_attribute(
            SpanAttributes.DB_NAME, self._db_api_integration.database
        )
        span.set_attribute(SpanAttributes.DB_STATEMENT, statement)

        for (
            attribute_key,
            attribute_value,
        ) in self._db_api_integration.span_attributes.items():
            span.set_attribute(attribute_key, attribute_value)

        if self._db_api_integration.capture_parameters and len(args) > 1:
            span.set_attribute("db.statement.parameters", str(args[1]))

    def get_operation_name(
        self, cursor: CursorT, args: tuple[Any, ...]
    ) -> str:  # pylint: disable=no-self-use
        if args and isinstance(args[0], str):
            # Strip leading comments so we get the operation name.
            return self._leading_comment_remover.sub("", args[0]).split()[0]
        return ""

    def get_statement(self, cursor: CursorT, args: tuple[Any, ...]):  # pylint: disable=no-self-use
        if not args:
            return ""
        statement = args[0]
        if isinstance(statement, bytes):
            return statement.decode("utf8", "replace")
        return statement

    def traced_execution(
        self,
        cursor: CursorT,
        query_method: Callable[..., Any],
        *args: tuple[Any, ...],
        **kwargs: dict[Any, Any],
    ):
        name = self.get_operation_name(cursor, args)
        if not name:
            name = (
                self._db_api_integration.database
                if self._db_api_integration.database
                else self._db_api_integration.name
            )

        with self._db_api_integration._tracer.start_as_current_span(
            name, kind=SpanKind.CLIENT
        ) as span:
            if span.is_recording():
                if args and self._commenter_enabled:
                    if self._enable_attribute_commenter:
                        # sqlcomment is added to executed query and db.statement span attribute
                        args = self._update_args_with_added_sql_comment(
                            args, cursor
                        )
                        self._populate_span(span, cursor, *args)
                    else:
                        # sqlcomment is only added to executed query
                        # so db.statement is set before add_sql_comment
                        self._populate_span(span, cursor, *args)
                        args = self._update_args_with_added_sql_comment(
                            args, cursor
                        )
                else:
                    # no sqlcomment anywhere
                    self._populate_span(span, cursor, *args)
            return query_method(*args, **kwargs)


# pylint: disable=abstract-method
class TracedCursorProxy(wrapt.ObjectProxy, Generic[CursorT]):
    # pylint: disable=unused-argument
    def __init__(
        self,
        cursor: CursorT,
        db_api_integration: DatabaseApiIntegration,
    ):
        wrapt.ObjectProxy.__init__(self, cursor)
        self._self_cursor_tracer = CursorTracer[CursorT](db_api_integration)

    def execute(self, *args: Any, **kwargs: Any):
        return self._self_cursor_tracer.traced_execution(
            self.__wrapped__, self.__wrapped__.execute, *args, **kwargs
        )

    def executemany(self, *args: Any, **kwargs: Any):
        return self._self_cursor_tracer.traced_execution(
            self.__wrapped__, self.__wrapped__.executemany, *args, **kwargs
        )

    def callproc(self, *args: Any, **kwargs: Any):
        return self._self_cursor_tracer.traced_execution(
            self.__wrapped__, self.__wrapped__.callproc, *args, **kwargs
        )

    def __enter__(self):
        self.__wrapped__.__enter__()
        return self

    def __exit__(self, *args, **kwargs):
        self.__wrapped__.__exit__(*args, **kwargs)


def get_traced_cursor_proxy(
    cursor: CursorT,
    db_api_integration: DatabaseApiIntegration,
    *args: Any,
    **kwargs: Any,
) -> TracedCursorProxy[CursorT]:
    return TracedCursorProxy(cursor, db_api_integration)