chore(sqlalchemy): Drop transaction-based tracing

sentry_sdk/integrations/asyncpg.py

@@ -2,16 +2,14 @@
 
 import contextlib
 import re
-from typing import Any, Awaitable, Callable, Iterator, TypeVar, Union
+from typing import Any, Awaitable, Callable, Iterator, TypeVar
 
 import sentry_sdk
 from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.traces import StreamedSpan
-from sentry_sdk.tracing import Span
 from sentry_sdk.tracing_utils import (
     add_query_source,
-    has_span_streaming_enabled,
     record_sql_queries,
 )
 from sentry_sdk.utils import (
@@ -95,11 +93,6 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
             span_origin=AsyncPGIntegration.origin,
         ) as span:
             res = await f(*args, **kwargs)
-            if isinstance(span, StreamedSpan):
-                with capture_internal_exceptions():
-                    add_query_source(span)
-
-        if not isinstance(span, StreamedSpan):
             with capture_internal_exceptions():
                 add_query_source(span)
 
@@ -118,7 +111,7 @@ def _record(
     params_list: "tuple[Any, ...] | None",
     *,
     executemany: bool = False,
-) -> "Iterator[Union[Span, StreamedSpan]]":
+) -> "Iterator[StreamedSpan]":
     client = sentry_sdk.get_client()
     integration = client.get_integration(AsyncPGIntegration)
     if integration is not None and not integration._record_params:
@@ -152,11 +145,6 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
 
             res = await f(*args, **kwargs)
 
-            if isinstance(span, StreamedSpan):
-                with capture_internal_exceptions():
-                    add_query_source(span)
-
-        if not isinstance(span, StreamedSpan):
             with capture_internal_exceptions():
                 add_query_source(span)
 
@@ -194,11 +182,6 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
             _set_db_data(span, cursor._connection)
             res = await f(*args, **kwargs)
 
-            if isinstance(span, StreamedSpan):
-                with capture_internal_exceptions():
-                    add_query_source(span)
-
-        if not isinstance(span, StreamedSpan):
             with capture_internal_exceptions():
                 add_query_source(span)
 
@@ -219,95 +202,51 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
         database = kwargs["params"].database
         addr = kwargs.get("addr")
 
-        if has_span_streaming_enabled(client.options):
-            span_attributes = {
-                "sentry.op": OP.DB,
-                "sentry.origin": AsyncPGIntegration.origin,
-                SPANDATA.DB_SYSTEM_NAME: "postgresql",
-                SPANDATA.DB_USER: user,
-                SPANDATA.DB_NAMESPACE: database,
-                SPANDATA.DB_DRIVER_NAME: "asyncpg",
-            }
-            if addr:
-                try:
-                    span_attributes[SPANDATA.SERVER_ADDRESS] = addr[0]
-                    span_attributes[SPANDATA.SERVER_PORT] = addr[1]
-                except IndexError:
-                    pass
+        span_attributes = {
+            "sentry.op": OP.DB,
+            "sentry.origin": AsyncPGIntegration.origin,
+            SPANDATA.DB_SYSTEM_NAME: "postgresql",
+            SPANDATA.DB_USER: user,
+            SPANDATA.DB_NAMESPACE: database,
+            SPANDATA.DB_DRIVER_NAME: "asyncpg",
+        }
+        if addr:
+            try:
+                span_attributes[SPANDATA.SERVER_ADDRESS] = addr[0]
+                span_attributes[SPANDATA.SERVER_PORT] = addr[1]
+            except IndexError:
+                pass
+
+        with capture_internal_exceptions():
+            sentry_sdk.add_breadcrumb(
+                message="connect", category="query", data=span_attributes
+            )
 
-            with capture_internal_exceptions():
-                sentry_sdk.add_breadcrumb(
-                    message="connect", category="query", data=span_attributes
-                )
-
-            if sentry_sdk.traces.get_current_span() is None:
-                return await f(*args, **kwargs)
-
-            with sentry_sdk.traces.start_span(
-                name="connect", attributes=span_attributes
-            ):
-                return await f(*args, **kwargs)
-
-        with sentry_sdk.start_span(
-            op=OP.DB,
-            name="connect",
-            origin=AsyncPGIntegration.origin,
-        ) as span:
-            span.set_data(SPANDATA.DB_SYSTEM, "postgresql")
-            if addr:
-                try:
-                    span.set_data(SPANDATA.SERVER_ADDRESS, addr[0])
-                    span.set_data(SPANDATA.SERVER_PORT, addr[1])
-                except IndexError:
-                    pass
-            span.set_data(SPANDATA.DB_NAME, database)
-            span.set_data(SPANDATA.DB_USER, user)
-            span.set_data(SPANDATA.DB_DRIVER_NAME, "asyncpg")
+        if sentry_sdk.traces.get_current_span() is None:
+            return await f(*args, **kwargs)
 
-            with capture_internal_exceptions():
-                sentry_sdk.add_breadcrumb(
-                    message="connect", category="query", data=span._data
-                )
+        with sentry_sdk.traces.start_span(name="connect", attributes=span_attributes):
             return await f(*args, **kwargs)
 
     return _inner
 
 
-def _set_db_data(span: "Union[Span, StreamedSpan]", conn: "Any") -> None:
+def _set_db_data(span: "StreamedSpan", conn: "Any") -> None:
     addr = conn._addr
     database = conn._params.database
     user = conn._params.user
 
-    if isinstance(span, StreamedSpan):
-        span.set_attribute(SPANDATA.DB_SYSTEM_NAME, "postgresql")
-        span.set_attribute(SPANDATA.DB_DRIVER_NAME, "asyncpg")
-        if addr:
-            try:
-                span.set_attribute(SPANDATA.SERVER_ADDRESS, addr[0])
-                span.set_attribute(SPANDATA.SERVER_PORT, addr[1])
-            except IndexError:
-                pass
-
-        if database:
-            span.set_attribute(SPANDATA.DB_NAMESPACE, database)
-
-        if user:
-            span.set_attribute(SPANDATA.DB_USER, user)
-    else:
-        # Remove this else block once we've completely migrated to streamed spans
-        # The use of deprecated attributes here is to ensure backwards compatibility
-        span.set_data(SPANDATA.DB_SYSTEM, "postgresql")
-        span.set_data(SPANDATA.DB_DRIVER_NAME, "asyncpg")
-
-        if addr:
-            try:
-                span.set_data(SPANDATA.SERVER_ADDRESS, addr[0])
-                span.set_data(SPANDATA.SERVER_PORT, addr[1])
-            except IndexError:
-                pass
+    span.set_attribute(SPANDATA.DB_SYSTEM_NAME, "postgresql")
+    span.set_attribute(SPANDATA.DB_DRIVER_NAME, "asyncpg")
+    if addr:
+        try:
+            span.set_attribute(SPANDATA.SERVER_ADDRESS, addr[0])
+            span.set_attribute(SPANDATA.SERVER_PORT, addr[1])
+        except IndexError:
+            pass
 
-        if database:
-            span.set_data(SPANDATA.DB_NAME, database)
+    if database:
+        span.set_attribute(SPANDATA.DB_NAMESPACE, database)
 
-        if user:
-            span.set_data(SPANDATA.DB_USER, user)
+    if user:
+        span.set_attribute(SPANDATA.DB_USER, user)

sentry_sdk/integrations/django/__init__.py

@@ -84,7 +84,6 @@
     )
     from sentry_sdk.integrations.wsgi import _ScopedResponse
     from sentry_sdk.traces import StreamedSpan
-    from sentry_sdk.tracing import Span
 
 
 TRANSACTION_STYLE_VALUES = ("function_name", "url")
@@ -845,15 +844,10 @@ def _rollback(self: "BaseDatabaseWrapper") -> None:
 
 
 def _set_db_data(
-    span: "Union[Span, StreamedSpan]",
+    span: "StreamedSpan",
     cursor_or_db: "Any",
     db_operation: "Optional[str]" = None,
 ) -> None:
-    # TODO: remove this once record_sql_queries drops support for old spans
-    from sentry_sdk.traces import StreamedSpan
-
-    if not isinstance(span, StreamedSpan):
-        return
     db = cursor_or_db.db if hasattr(cursor_or_db, "db") else cursor_or_db
     vendor = db.vendor
     span.set_attribute(SPANDATA.DB_SYSTEM_NAME, vendor)

sentry_sdk/integrations/sqlalchemy.py

@@ -1,7 +1,6 @@
-from sentry_sdk.consts import SPANDATA, SPANSTATUS
+from sentry_sdk.consts import SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.traces import SpanStatus, StreamedSpan
-from sentry_sdk.tracing import Span
 from sentry_sdk.tracing_utils import (
     add_query_source,
     record_sql_queries,
@@ -22,7 +21,7 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from typing import Any, ContextManager, Optional, Union
+    from typing import Any, ContextManager, Optional
 
 
 class SqlalchemyIntegration(Integration):
@@ -79,34 +78,27 @@ def _after_cursor_execute(
     )
 
     # Record query source immediately before span is finished: accurate end timestamp and before the span is flushed.
-    span: "Optional[Union[Span, StreamedSpan]]" = getattr(
-        context, "_sentry_sql_span", None
-    )
-    if isinstance(span, StreamedSpan):
+    span: "Optional[StreamedSpan]" = getattr(context, "_sentry_sql_span", None)
+    if span is not None:
         with capture_internal_exceptions():
             add_query_source(span)
 
     if ctx_mgr is not None:
         context._sentry_sql_span_manager = None
         ctx_mgr.__exit__(None, None, None)
 
-    if isinstance(span, Span):
-        with capture_internal_exceptions():
-            add_query_source(span)
-
 
 def _handle_error(context: "Any", *args: "Any") -> None:
     execution_context = context.execution_context
     if execution_context is None:
         return
 
-    span: "Optional[Span]" = getattr(execution_context, "_sentry_sql_span", None)
+    span: "Optional[StreamedSpan]" = getattr(
+        execution_context, "_sentry_sql_span", None
+    )
 
     if span is not None:
-        if isinstance(span, StreamedSpan):
-            span.status = SpanStatus.ERROR
-        else:
-            span.set_status(SPANSTATUS.INTERNAL_ERROR)
+        span.status = SpanStatus.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
@@ -142,43 +134,30 @@ def _get_db_system(name: str) -> "Optional[str]":
     return None
 
 
-def _set_db_data(span: "Union[Span, StreamedSpan]", conn: "Any") -> None:
+def _set_db_data(span: "StreamedSpan", conn: "Any") -> None:
     db_system = _get_db_system(conn.engine.name)
 
-    if isinstance(span, StreamedSpan):
-        if db_system is not None:
-            span.set_attribute(SPANDATA.DB_SYSTEM_NAME, db_system)
-    else:
-        if db_system is not None:
-            span.set_data(SPANDATA.DB_SYSTEM, db_system)
-
-    if isinstance(span, StreamedSpan):
-        set_on_span = span.set_attribute
-    else:
-        set_on_span = span.set_data
+    if db_system is not None:
+        span.set_attribute(SPANDATA.DB_SYSTEM_NAME, db_system)
 
     try:
         driver = conn.dialect.driver
         if driver:
-            set_on_span(SPANDATA.DB_DRIVER_NAME, driver)
+            span.set_attribute(SPANDATA.DB_DRIVER_NAME, driver)
     except Exception:
         pass
 
     if conn.engine.url is None:
         return
 
     db_name = conn.engine.url.database
-    if isinstance(span, StreamedSpan):
-        if db_name is not None:
-            span.set_attribute(SPANDATA.DB_NAMESPACE, db_name)
-    else:
-        if db_name is not None:
-            span.set_data(SPANDATA.DB_NAME, db_name)
+    if db_name is not None:
+        span.set_attribute(SPANDATA.DB_NAMESPACE, db_name)
 
     server_address = conn.engine.url.host
     if server_address is not None:
-        set_on_span(SPANDATA.SERVER_ADDRESS, server_address)
+        span.set_attribute(SPANDATA.SERVER_ADDRESS, server_address)
 
     server_port = conn.engine.url.port
     if server_port is not None:
-        set_on_span(SPANDATA.SERVER_PORT, server_port)
+        span.set_attribute(SPANDATA.SERVER_PORT, server_port)