opentelemetry-instrumentation-dbapi: implement proper handling of t-string queries

.changelog/4697.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-dbapi`: implement proper handling of t-string queries

instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py

@@ -1,5 +1,6 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
+# pylint: disable=too-many-lines
 
 """
 The trace integration with Database API supports libraries that follow the
@@ -160,8 +161,10 @@
 import functools
 import logging
 import re
+import sys
 import time
-from typing import Any, Awaitable, Callable, Generic, TypeVar
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Generic, TypeVar
 
 from wrapt import wrap_function_wrapper
 
@@ -213,6 +216,20 @@
 from opentelemetry.trace import SpanKind, TracerProvider, get_tracer
 from opentelemetry.util._importlib_metadata import version as util_version
 
+if sys.version_info >= (3, 14):
+    from string.templatelib import Template as _Template
+else:
+    _Template = ()
+
+if TYPE_CHECKING:
+    if sys.version_info >= (3, 14):
+        from string.templatelib import Template
+    else:
+        from typing import Never
+
+        Template = Never
+
+
 _DB_DRIVER_ALIASES = {
     "MySQLdb": "mysqlclient",
 }
@@ -682,6 +699,16 @@ def get_traced_connection_proxy(
     return TracedConnectionProxy(connection, db_api_integration)
 
 
+def _t_string_to_str(template: Template) -> str:
+    """Render a PEP 750 Template as a string with expression placeholders."""
+    parts: list[str] = []
+    for idx, literal in enumerate(template.strings):
+        parts.append(literal)
+        if idx < len(template.interpolations):
+            parts.append(f"{{{template.interpolations[idx].expression}}}")
+    return "".join(parts)
+
+
 class CursorTracer(Generic[CursorT]):
     def __init__(self, db_api_integration: DatabaseApiIntegration) -> None:
         self._db_api_integration = db_api_integration
@@ -742,13 +769,14 @@ def _update_args_with_added_sql_comment(self, args, cursor) -> tuple:
             args_list = list(args)
             self._capture_mysql_version(cursor)
             commenter_data = self._get_commenter_data()
-            # Convert sql statement to string, handling  psycopg2.sql.Composable object
-            if hasattr(args_list[0], "as_string"):
-                args_list[0] = args_list[0].as_string(cursor.connection)
-
-            args_list[0] = str(args_list[0])
-            statement = _add_sql_comment(args_list[0], **commenter_data)
-            args_list[0] = statement
+            if not isinstance(args_list[0], str) and not isinstance(
+                args_list[0], _Template
+            ):
+                # Convert sql statement to string, handling psycopg2.sql.Composable object
+                if hasattr(args_list[0], "as_string"):
+                    args_list[0] = args_list[0].as_string(cursor.connection)
+                args_list[0] = str(args_list[0])
+            args_list[0] = _add_sql_comment(args_list[0], **commenter_data)
             args = tuple(args_list)
         except Exception as exc:  # pylint: disable=broad-except
             _logger.exception(
@@ -792,21 +820,34 @@ def _populate_span(
         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):
+    def get_operation_name(self, cursor: CursorT, args: Sequence[Any]) -> str:  # pylint: disable=no-self-use
+        if not args:
+            return ""
+        query = args[0]
+        if isinstance(query, _Template):
+            first_literal = query.strings[0] if query.strings else ""
+            tokens = self._leading_comment_remover.sub(
+                "", first_literal
+            ).split()
+            return tokens[0] if tokens else ""
+
+        # `query` can be an empty string. See #2643
+        if query and isinstance(query, str):
             # Strip leading comments so we get the operation name.
-            return self._leading_comment_remover.sub("", args[0]).split()[0]
+            return self._leading_comment_remover.sub("", query).split()[0]
         return ""
 
-    def get_statement(self, cursor: CursorT, args: tuple[Any, ...]):  # pylint: disable=no-self-use
+    def get_statement(self, cursor: CursorT, args: Sequence[Any]) -> str:  # pylint: disable=no-self-use
         if not args:
             return ""
         statement = args[0]
+        if isinstance(statement, _Template):
+            return _t_string_to_str(statement)
         if isinstance(statement, bytes):
             return statement.decode("utf8", "replace")
-        return statement
+        if isinstance(statement, str):
+            return statement
+        return ""
 
     def _get_metric_attributes(
         self,

instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py

@@ -7,8 +7,11 @@
 import contextlib
 import logging
 import re
+import sys
 from unittest import mock
 
+import pytest
+
 from opentelemetry import context
 from opentelemetry import trace as trace_api
 from opentelemetry.instrumentation import dbapi
@@ -1728,6 +1731,86 @@ def test_capture_mysql_version_fallback(self, mock_logger):
             db_integration.commenter_data["mysql_client_version"], "unknown"
         )
 
+    @pytest.mark.skipif(
+        sys.version_info < (3, 14),
+        reason="requires Python 3.14+ for t-strings",
+    )
+    def test_t_string_span_attributes(self):
+        # pylint: disable-next=import-outside-toplevel,no-name-in-module
+        from string.templatelib import (  # noqa: PLC0415
+            Interpolation,
+            Template,
+        )
+
+        connection_props = _get_default_connection_props()
+        connection_attributes = _get_default_connection_attributes()
+        db_integration = dbapi.DatabaseApiIntegration(
+            "instrumenting_module_test_name",
+            "testcomponent",
+            connection_attributes,
+        )
+        mock_connection = db_integration.wrapped_connection(
+            mock_connect, {}, connection_props
+        )
+        cursor = mock_connection.cursor()
+        value = 42
+        template = Template(
+            "SELECT ", Interpolation(value, "value"), " FROM foo"
+        )
+        cursor.execute(template)
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(
+            span.attributes[DB_STATEMENT], "SELECT {value} FROM foo"
+        )
+
+    @pytest.mark.skipif(
+        sys.version_info < (3, 14),
+        reason="requires Python 3.14+ for t-strings",
+    )
+    def test_t_string_commenter(self):
+        # pylint: disable-next=import-outside-toplevel,no-name-in-module
+        from string.templatelib import Interpolation, Template  # noqa: PLC0415
+
+        connect_module = mock.MagicMock()
+        connect_module.__name__ = "test"
+        connect_module.__version__ = mock.MagicMock()
+        connect_module.pq.version.return_value = 123
+        connect_module.apilevel = 123
+        connect_module.threadsafety = 123
+        connect_module.paramstyle = "test"
+        db_integration = dbapi.DatabaseApiIntegration(
+            "instrumenting_module_test_name",
+            "postgresql",
+            enable_commenter=True,
+            commenter_options={"db_driver": False, "dbapi_level": False},
+            connect_module=connect_module,
+        )
+        mock_connection = db_integration.wrapped_connection(
+            mock_connect, {}, {}
+        )
+        cursor = mock_connection.cursor()
+        value = 42
+        template = Template(
+            "SELECT ", Interpolation(value, "value"), " FROM foo"
+        )
+        cursor.executemany(template)
+        self.assertIsInstance(cursor.query, Template)
+        self.assertRegex(
+            cursor.query.strings[-1],
+            r" FROM foo /\*dbapi_threadsafety=123,driver_paramstyle='test',"
+            r"libpq_version=123,traceparent='\d{1,2}-[a-zA-Z0-9_]{32}-[a-zA-Z0-9_]{16}-\d{1,2}'\*/",
+        )
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(
+            span.attributes[DB_STATEMENT], "SELECT {value} FROM foo"
+        )
+
 
 # pylint: disable=unused-argument
 def mock_connect(*args, **kwargs):

instrumentation/opentelemetry-instrumentation-psycopg/src/opentelemetry/instrumentation/psycopg/__init__.py

@@ -329,22 +329,21 @@ def get_operation_name(self, cursor: CursorT, args: list[Any]) -> str:
         statement = args[0]
         if isinstance(statement, Composable):
             statement = statement.as_string(cursor)
-
-        # `statement` can be empty string. See #2643
-        if statement and isinstance(statement, str):
-            # Strip leading comments so we get the operation name.
-            return self._leading_comment_remover.sub("", statement).split()[0]
-
-        return ""
+            return (
+                self._leading_comment_remover.sub("", statement).split()[0]
+                if statement
+                else ""
+            )
+        return super().get_operation_name(cursor, args)
 
     def get_statement(self, cursor: CursorT, args: list[Any]) -> str:
         if not args:
             return ""
 
         statement = args[0]
         if isinstance(statement, Composable):
-            statement = statement.as_string(cursor)
-        return statement
+            return statement.as_string(cursor)
+        return super().get_statement(cursor, args)
 
 
 def _new_cursor_factory(

instrumentation/opentelemetry-instrumentation-psycopg/test-requirements-1.txt

@@ -3,7 +3,7 @@ Deprecated==1.2.14
 iniconfig==2.0.0
 packaging==24.0
 pluggy==1.6.0
-psycopg==3.2.2
+psycopg==3.3.4
 py-cpuinfo==9.0.0
 pytest==7.4.4
 tomli==2.0.1

instrumentation/opentelemetry-instrumentation-psycopg/tests/test_psycopg_integration.py

@@ -1,11 +1,14 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
+# pylint: disable=no-member  # psycopg stubs reference string.templatelib on 3.14+
 
 import asyncio
+import sys
 import types
 from unittest import IsolatedAsyncioTestCase, mock
 
 import psycopg
+import pytest
 from psycopg.sql import SQL, Composed
 
 import opentelemetry.instrumentation.psycopg
@@ -450,6 +453,61 @@ def test_sqlcommenter_disabled(self, event_mocked):
         kwargs = event_mocked.call_args[1]
         self.assertEqual(kwargs["enable_commenter"], False)
 
+    @pytest.mark.skipif(
+        sys.version_info < (3, 14),
+        reason="requires Python 3.14+ for t-strings",
+    )
+    def test_t_string_span_attributes(self):
+        # pylint: disable-next=import-outside-toplevel,no-name-in-module,no-member
+        from string.templatelib import Interpolation, Template  # noqa: PLC0415
+
+        PsycopgInstrumentor().instrument()
+        cnx = psycopg.connect(database="test")
+        cursor = cnx.cursor()
+        value = 42
+        template = Template(
+            "SELECT ", Interpolation(value, "value"), " FROM foo"
+        )
+        cursor.execute(template)
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(
+            span.attributes["db.statement"], "SELECT {value} FROM foo"
+        )
+
+    @pytest.mark.skipif(
+        sys.version_info < (3, 14),
+        reason="requires Python 3.14+ for t-strings",
+    )
+    def test_t_string_commenter(self):
+        # pylint: disable-next=import-outside-toplevel,no-name-in-module,no-member
+        from string.templatelib import Interpolation, Template  # noqa: PLC0415
+
+        PsycopgInstrumentor().instrument(enable_commenter=True)
+        cnx = psycopg.connect(database="test")
+        cursor = cnx.cursor()
+        MockCursor.executemany.reset_mock()
+        value = 42
+        template = Template(
+            "SELECT ", Interpolation(value, "value"), " FROM foo;"
+        )
+        cursor.executemany(template)
+        called_query = MockCursor.executemany.call_args[0][0]
+        self.assertIsInstance(called_query, Template)
+        self.assertRegex(
+            called_query.strings[-1],
+            r" FROM foo /\*.*traceparent=.*\*/;",
+        )
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(
+            span.attributes["db.statement"], "SELECT {value} FROM foo;"
+        )
+
 
 class TestPostgresqlIntegrationAsync(
     PostgresqlIntegrationTestMixin, TestBase, IsolatedAsyncioTestCase
@@ -634,3 +692,28 @@ async def test_instrument_connection_uses_async_cursor_factory(self):
         self.assertEqualSpanInstrumentationScope(
             span, opentelemetry.instrumentation.psycopg
         )
+
+    @pytest.mark.skipif(
+        sys.version_info < (3, 14),
+        reason="requires Python 3.14+ for t-strings",
+    )
+    async def test_t_string_span_attributes_async(self):
+        # pylint: disable-next=import-outside-toplevel,no-name-in-module,no-member
+        from string.templatelib import Interpolation, Template  # noqa: PLC0415
+
+        PsycopgInstrumentor().instrument()
+        cnx = await psycopg.AsyncConnection.connect("test")
+        self.assertTrue(issubclass(cnx.cursor_factory, MockAsyncCursor))
+        async with cnx.cursor() as cursor:
+            value = 42
+            template = Template(
+                "SELECT ", Interpolation(value, "value"), " FROM foo"
+            )
+            await cursor.execute(template)
+        spans_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+        self.assertEqual(span.name, "SELECT")
+        self.assertEqual(
+            span.attributes["db.statement"], "SELECT {value} FROM foo"
+        )