diff --git a/sentry_sdk/integrations/asyncpg.py b/sentry_sdk/integrations/asyncpg.py index c746c3a8f6..e99560eecf 100644 --- a/sentry_sdk/integrations/asyncpg.py +++ b/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) diff --git a/sentry_sdk/integrations/django/__init__.py b/sentry_sdk/integrations/django/__init__.py index c158c5e66d..dafc152071 100644 --- a/sentry_sdk/integrations/django/__init__.py +++ b/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) diff --git a/sentry_sdk/integrations/sqlalchemy.py b/sentry_sdk/integrations/sqlalchemy.py index 8878ea4dac..1d7136ec53 100644 --- a/sentry_sdk/integrations/sqlalchemy.py +++ b/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,10 +78,8 @@ 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) @@ -90,23 +87,18 @@ def _after_cursor_execute( 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,25 +134,16 @@ 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 @@ -168,17 +151,13 @@ def _set_db_data(span: "Union[Span, StreamedSpan]", conn: "Any") -> 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) diff --git a/sentry_sdk/tracing_utils.py b/sentry_sdk/tracing_utils.py index fa26876fe0..888668368d 100644 --- a/sentry_sdk/tracing_utils.py +++ b/sentry_sdk/tracing_utils.py @@ -134,10 +134,8 @@ def record_sql_queries( record_cursor_repr: bool = False, span_origin: str = "manual", span_op_override_value: "Optional[str]" = None, -) -> "Generator[Union[sentry_sdk.tracing.Span, sentry_sdk.traces.StreamedSpan], None, None]": +) -> "Generator[sentry_sdk.traces.StreamedSpan, None, None]": # TODO: Bring back capturing of params by default - # TODO: Once we drop span streaming from this, remove the hack from django's - # _set_db_data client = sentry_sdk.get_client() if has_data_collection_enabled(client.options): if client.options["data_collection"]["database_query_data"]: @@ -176,31 +174,19 @@ def record_sql_queries( with capture_internal_exceptions(): sentry_sdk.add_breadcrumb(message=query, category="query", data=data) - if has_span_streaming_enabled(client.options): - additional_attributes = {} - if query is not None: - additional_attributes["db.query.text"] = query - - with sentry_sdk.traces.start_span( - name="" if query is None else query, - attributes={ - "sentry.origin": span_origin, - "sentry.op": span_op_override_value - if span_op_override_value - else OP.DB, - **additional_attributes, - }, - ) as span: - yield span - else: - with sentry_sdk.start_span( - op=span_op_override_value if span_op_override_value is not None else OP.DB, - name=query, - origin=span_origin, - ) as span: - for k, v in data.items(): - span.set_data(k, v) - yield span + additional_attributes = {} + if query is not None: + additional_attributes["db.query.text"] = query + + with sentry_sdk.traces.start_span( + name="" if query is None else query, + attributes={ + "sentry.origin": span_origin, + "sentry.op": span_op_override_value if span_op_override_value else OP.DB, + **additional_attributes, + }, + ) as span: + yield span def add_http_breadcrumb(status_code: "Optional[int]", data: "dict[str, Any]") -> None: @@ -332,7 +318,7 @@ def add_source( def add_query_source( - span: "Union[sentry_sdk.tracing.Span, sentry_sdk.traces.StreamedSpan]", + span: "sentry_sdk.traces.StreamedSpan", ) -> None: """ Adds OTel compatible source code information to a database query span @@ -341,13 +327,6 @@ def add_query_source( if not client.is_active(): return - if isinstance(span, Span): - # In the StreamedSpan case, we need to add the extra span information before - # the span finishes, so it's expected that this will be None. In the Span case, - # it should already be finished. - if span.timestamp is None: - return - if span.start_timestamp is None: return @@ -355,12 +334,7 @@ def add_query_source( if not should_add_query_source: return - if isinstance(span, StreamedSpan): - end_timestamp = span.end_timestamp - else: - end_timestamp = span.timestamp - - end_timestamp = end_timestamp or datetime.now(timezone.utc) + end_timestamp = span.end_timestamp or datetime.now(timezone.utc) duration = end_timestamp - span.start_timestamp threshold = client.options.get("db_query_source_threshold_ms", 0) diff --git a/tests/integrations/asyncpg/test_asyncpg.py b/tests/integrations/asyncpg/test_asyncpg.py index 1991b9af05..23b3058dcf 100644 --- a/tests/integrations/asyncpg/test_asyncpg.py +++ b/tests/integrations/asyncpg/test_asyncpg.py @@ -12,6 +12,7 @@ import datetime import os from contextlib import contextmanager +from datetime import timezone from unittest import mock import asyncpg @@ -20,7 +21,7 @@ from asyncpg import Connection, connect import sentry_sdk -from sentry_sdk import capture_message, start_transaction +from sentry_sdk import capture_message from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations.asyncpg import AsyncPGIntegration from sentry_sdk.tracing_utils import record_sql_queries @@ -44,21 +45,6 @@ def _get_db_name(): PG_USER, PG_PASSWORD, PG_HOST, PG_NAME ) CRUMBS_CONNECT = { - "category": "query", - "data": ApproxDict( - { - "db.name": PG_NAME, - "db.system": "postgresql", - "db.user": PG_USER, - "db.driver.name": "asyncpg", - "server.address": PG_HOST, - "server.port": PG_PORT, - } - ), - "message": "connect", - "type": "default", -} -CRUMBS_CONNECT_STREAMING = { "category": "query", "data": ApproxDict( { @@ -110,22 +96,19 @@ async def _clean_pg(): @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_connect( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ) -> None: sentry_init( integrations=[AsyncPGIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, ) - if span_streaming: - items = capture_items("event") - else: - events = capture_events() + items = capture_items("event") conn: Connection = await connect(PG_CONNECTION_URI) @@ -133,37 +116,29 @@ async def test_connect( capture_message("hi") - if span_streaming: - event = items[0].payload - else: - (event,) = events + event = items[0].payload for crumb in event["breadcrumbs"]["values"]: del crumb["timestamp"] - expected_crumbs_connect = ( - CRUMBS_CONNECT_STREAMING if span_streaming else CRUMBS_CONNECT - ) + expected_crumbs_connect = CRUMBS_CONNECT assert event["breadcrumbs"]["values"] == [expected_crumbs_connect] @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_execute( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ) -> None: sentry_init( integrations=[AsyncPGIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, ) - if span_streaming: - items = capture_items("event") - else: - events = capture_events() + items = capture_items("event") conn: Connection = await connect(PG_CONNECTION_URI) @@ -188,17 +163,12 @@ async def test_execute( capture_message("hi") - if span_streaming: - event = items[0].payload - else: - (event,) = events + event = items[0].payload for crumb in event["breadcrumbs"]["values"]: del crumb["timestamp"] - expected_crumbs_connect = ( - CRUMBS_CONNECT_STREAMING if span_streaming else CRUMBS_CONNECT - ) + expected_crumbs_connect = CRUMBS_CONNECT assert event["breadcrumbs"]["values"] == [ expected_crumbs_connect, { @@ -229,22 +199,19 @@ async def test_execute( @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_execute_many( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ) -> None: sentry_init( integrations=[AsyncPGIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, ) - if span_streaming: - items = capture_items("event") - else: - events = capture_events() + items = capture_items("event") conn: Connection = await connect(PG_CONNECTION_URI) @@ -260,19 +227,13 @@ async def test_execute_many( capture_message("hi") - if span_streaming: - event = items[0].payload - else: - (event,) = events + event = items[0].payload for crumb in event["breadcrumbs"]["values"]: del crumb["timestamp"] - expected_crumbs_connect = ( - CRUMBS_CONNECT_STREAMING if span_streaming else CRUMBS_CONNECT - ) assert event["breadcrumbs"]["values"] == [ - expected_crumbs_connect, + CRUMBS_CONNECT, { "category": "query", "data": {"db.executemany": True}, @@ -287,6 +248,7 @@ async def test_record_params(sentry_init, capture_events) -> None: sentry_init( integrations=[AsyncPGIntegration(record_params=True)], _experiments={"record_sql_params": True}, + trace_lifecycle="stream", ) events = capture_events() @@ -327,6 +289,7 @@ async def test_cursor(sentry_init, capture_events) -> None: sentry_init( integrations=[AsyncPGIntegration()], _experiments={"record_sql_params": True}, + trace_lifecycle="stream", ) events = capture_events() @@ -346,7 +309,7 @@ async def test_cursor(sentry_init, capture_events) -> None: async for record in conn.cursor( "SELECT * FROM users WHERE dob > $1", datetime.date(1970, 1, 1) ): - print(record) + pass await conn.close() @@ -381,6 +344,7 @@ async def test_cursor_manual(sentry_init, capture_events) -> None: sentry_init( integrations=[AsyncPGIntegration()], _experiments={"record_sql_params": True}, + trace_lifecycle="stream", ) events = capture_events() @@ -400,11 +364,9 @@ async def test_cursor_manual(sentry_init, capture_events) -> None: cur = await conn.cursor( "SELECT * FROM users WHERE dob > $1", datetime.date(1970, 1, 1) ) - record = await cur.fetchrow() - print(record) + await cur.fetchrow() while await cur.forward(1): - record = await cur.fetchrow() - print(record) + await cur.fetchrow() await conn.close() @@ -445,6 +407,7 @@ async def test_prepared_stmt(sentry_init, capture_events) -> None: sentry_init( integrations=[AsyncPGIntegration()], _experiments={"record_sql_params": True}, + trace_lifecycle="stream", ) events = capture_events() @@ -460,8 +423,8 @@ async def test_prepared_stmt(sentry_init, capture_events) -> None: stmt = await conn.prepare("SELECT * FROM users WHERE name = $1") - print(await stmt.fetchval("Bob")) - print(await stmt.fetchval("Alice")) + await stmt.fetchval("Bob") + await stmt.fetchval("Alice") await conn.close() @@ -494,6 +457,7 @@ async def test_connection_pool(sentry_init, capture_events) -> None: sentry_init( integrations=[AsyncPGIntegration()], _experiments={"record_sql_params": True}, + trace_lifecycle="stream", ) events = capture_events() @@ -555,60 +519,42 @@ async def test_connection_pool(sentry_init, capture_events) -> None: @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_disabled( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): - sentry_options = { - "integrations": [AsyncPGIntegration()], - "traces_sample_rate": 1.0, - "enable_db_query_source": False, - "db_query_source_threshold_ms": 0, - "trace_lifecycle": "stream" if span_streaming else "static", - } - - sentry_init(**sentry_options) - - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) - - await conn.close() - sentry_sdk.flush() + sentry_init( + integrations=[AsyncPGIntegration()], + traces_sample_rate=1.0, + enable_db_query_source=False, + db_query_source_threshold_ms=0, + trace_lifecycle="stream", + ) - spans = [item.payload for item in items] + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - assert len(spans) == 3 + await conn.execute( + "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", + ) - connect_span = spans[0] - insert_span = spans[1] - segment = spans[2] + await conn.close() - assert segment["name"] == "test_segment" - assert insert_span["name"].startswith("INSERT INTO") - assert connect_span["name"] == "connect" - data = insert_span.get("attributes", {}) - else: - events = capture_events() + sentry_sdk.flush() - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) + spans = [item.payload for item in items] - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) + assert len(spans) == 3 - await conn.close() + connect_span = spans[0] + insert_span = spans[1] + segment = spans[2] - (event,) = events - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") - data = span.get("data", {}) + assert segment["name"] == "test_segment" + assert insert_span["name"].startswith("INSERT INTO") + assert connect_span["name"] == "connect" + data = insert_span.get("attributes", {}) assert SPANDATA.CODE_LINENO not in data assert SPANDATA.CODE_NAMESPACE not in data @@ -618,148 +564,84 @@ async def test_query_source_disabled( @pytest.mark.asyncio @pytest.mark.parametrize("enable_db_query_source", [None, True]) -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_enabled( - sentry_init, capture_events, capture_items, enable_db_query_source, span_streaming + sentry_init, + capture_items, + enable_db_query_source, ): sentry_options = { "integrations": [AsyncPGIntegration()], "traces_sample_rate": 1.0, "db_query_source_threshold_ms": 0, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } if enable_db_query_source is not None: sentry_options["enable_db_query_source"] = enable_db_query_source sentry_init(**sentry_options) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) - - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - assert len(spans) == 3 + await conn.execute( + "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", + ) - connect_span = spans[0] - insert_span = spans[1] - segment = spans[2] + await conn.close() - assert segment["name"] == "test_segment" - assert insert_span["name"].startswith("INSERT INTO") - assert connect_span["name"] == "connect" - data = insert_span.get("attributes", {}) - else: - events = capture_events() + sentry_sdk.flush() - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) + spans = [item.payload for item in items] - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) + assert len(spans) == 3 - await conn.close() + connect_span = spans[0] + insert_span = spans[1] + segment = spans[2] - (event,) = events - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") - data = span.get("data", {}) + assert segment["name"] == "test_segment" + assert insert_span["name"].startswith("INSERT INTO") + assert connect_span["name"] == "connect" - lineno_key = "code.line.number" if span_streaming else SPANDATA.CODE_LINENO - filepath_key = "code.file.path" if span_streaming else SPANDATA.CODE_FILEPATH - - assert lineno_key in data - assert filepath_key in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FUNCTION in data + _assert_query_source(insert_span, "test_query_source_enabled") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) -async def test_query_source(sentry_init, capture_events, capture_items, span_streaming): +async def test_query_source(sentry_init, capture_items): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) - - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - - assert len(spans) == 3 - - connect_span = spans[0] - insert_span = spans[1] - segment = spans[2] - - assert segment["name"] == "test_segment" - assert insert_span["name"].startswith("INSERT INTO") - assert connect_span["name"] == "connect" - data = insert_span.get("attributes", {}) - else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - await conn.close() + await conn.execute( + "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", + ) - (event,) = events - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") - data = span.get("data", {}) + await conn.close() - lineno_key = "code.line.number" if span_streaming else SPANDATA.CODE_LINENO - filepath_key = "code.file.path" if span_streaming else SPANDATA.CODE_FILEPATH + sentry_sdk.flush() - assert lineno_key in data - assert filepath_key in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FUNCTION in data + spans = [item.payload for item in items] - assert type(data.get(lineno_key)) == int - assert data.get(lineno_key) > 0 - assert ( - data.get(SPANDATA.CODE_NAMESPACE) == "tests.integrations.asyncpg.test_asyncpg" - ) - assert data.get(filepath_key).endswith("tests/integrations/asyncpg/test_asyncpg.py") + assert len(spans) == 3 - is_relative_path = data.get(filepath_key)[0] != os.sep - assert is_relative_path + _, insert_span, _ = spans - assert data.get(SPANDATA.CODE_FUNCTION) == "test_query_source" + _assert_query_source(insert_span, "test_query_source") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_with_module_in_search_path( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): """ Test that query source is relative to the path of the module it ran in @@ -769,156 +651,90 @@ async def test_query_source_with_module_in_search_path( traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) from asyncpg_helpers.helpers import execute_query_in_connection - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await execute_query_in_connection( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - conn, - ) - - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - - assert len(spans) == 3 - - connect_span = spans[0] - insert_span = spans[1] - segment = spans[2] - - assert segment["name"] == "test_segment" - assert insert_span["name"].startswith("INSERT INTO") - assert connect_span["name"] == "connect" - data = insert_span.get("attributes", {}) - else: - events = capture_events() + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) + await execute_query_in_connection( + "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", + conn, + ) - await execute_query_in_connection( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - conn, - ) + await conn.close() + sentry_sdk.flush() - await conn.close() + spans = [item.payload for item in items] - (event,) = events + assert len(spans) == 3 - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") - data = span.get("data", {}) + connect_span = spans[0] + insert_span = spans[1] + segment = spans[2] - lineno_key = "code.line.number" if span_streaming else SPANDATA.CODE_LINENO - filepath_key = "code.file.path" if span_streaming else SPANDATA.CODE_FILEPATH + assert segment["name"] == "test_segment" + assert insert_span["name"].startswith("INSERT INTO") + assert connect_span["name"] == "connect" + data = insert_span.get("attributes", {}) - assert lineno_key in data - assert filepath_key in data + assert "code.line.number" in data + assert "code.file.path" in data assert SPANDATA.CODE_NAMESPACE in data assert SPANDATA.CODE_FUNCTION in data - assert type(data.get(lineno_key)) == int - assert data.get(lineno_key) > 0 - assert data.get(filepath_key) == "asyncpg_helpers/helpers.py" + assert type(data.get("code.line.number")) == int + assert data.get("code.line.number") > 0 + assert data.get("code.file.path") == "asyncpg_helpers/helpers.py" assert data.get(SPANDATA.CODE_NAMESPACE) == "asyncpg_helpers.helpers" - is_relative_path = data.get(filepath_key)[0] != os.sep + is_relative_path = data.get("code.file.path")[0] != os.sep assert is_relative_path assert data.get(SPANDATA.CODE_FUNCTION) == "execute_query_in_connection" @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_no_query_source_if_duration_too_short( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, - db_query_source_threshold_ms=100, - trace_lifecycle="stream" if span_streaming else "static", + db_query_source_threshold_ms=100000, + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - - @contextmanager - def fake_record_sql_queries_streaming(*args, **kwargs): - with record_sql_queries(*args, **kwargs) as span: - pass - span._start_timestamp = datetime.datetime(2024, 1, 1, microsecond=0) - if span_streaming: - span._end_timestamp = datetime.datetime(2024, 1, 1, microsecond=99999) - else: - span._timestamp = datetime.datetime(2024, 1, 1, microsecond=99999) - yield span - - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - with mock.patch( - "sentry_sdk.integrations.asyncpg.record_sql_queries", - fake_record_sql_queries_streaming, - ): - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) - - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - - assert len(spans) == 3 + items = capture_items("span") - connect_span = spans[0] - insert_span = spans[1] - segment = spans[2] + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - assert segment["name"] == "test_segment" - assert insert_span["name"].startswith("INSERT INTO") - assert connect_span["name"] == "connect" - data = insert_span.get("attributes", {}) - else: - events = capture_events() + await conn.execute( + "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", + ) - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) + await conn.close() - @contextmanager - def fake_record_sql_queries(*args, **kwargs): - with record_sql_queries(*args, **kwargs) as span: - pass - span.start_timestamp = datetime.datetime(2024, 1, 1, microsecond=0) - span.timestamp = datetime.datetime(2024, 1, 1, microsecond=99999) - yield span + sentry_sdk.flush() - with mock.patch( - "sentry_sdk.integrations.asyncpg.record_sql_queries", - fake_record_sql_queries, - ): - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES ('Alice', 'secret', '1990-12-25')", - ) + spans = [item.payload for item in items] - await conn.close() + assert len(spans) == 3 - (event,) = events + connect_span = spans[0] + insert_span = spans[1] + segment = spans[2] - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") - data = span.get("data", {}) + assert segment["name"] == "test_segment" + assert insert_span["name"].startswith("INSERT INTO") + assert connect_span["name"] == "connect" + data = insert_span.get("attributes", {}) assert SPANDATA.CODE_LINENO not in data assert SPANDATA.CODE_NAMESPACE not in data @@ -927,26 +743,27 @@ def fake_record_sql_queries(*args, **kwargs): @pytest.mark.asyncio -async def test_query_source_if_duration_over_threshold(sentry_init, capture_events): +async def test_query_source_if_duration_over_threshold(sentry_init, capture_items): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, + trace_lifecycle="stream", enable_db_query_source=True, db_query_source_threshold_ms=100, ) - events = capture_events() + items = capture_items() - with start_transaction(name="test_transaction", sampled=True): + with sentry_sdk.traces.start_span(name="test_segment"): conn: Connection = await connect(PG_CONNECTION_URI) @contextmanager def fake_record_sql_queries(*args, **kwargs): with record_sql_queries(*args, **kwargs) as span: - pass - span.start_timestamp = datetime.datetime(2024, 1, 1, microsecond=0) - span.timestamp = datetime.datetime(2024, 1, 1, microsecond=100001) - yield span + span._start_timestamp = datetime.datetime( + 2024, 1, 1, microsecond=0, tzinfo=timezone.utc + ) + yield span with mock.patch( "sentry_sdk.integrations.asyncpg.record_sql_queries", @@ -958,181 +775,73 @@ def fake_record_sql_queries(*args, **kwargs): await conn.close() - (event,) = events - - span = event["spans"][-1] - assert span["description"].startswith("INSERT INTO") + sentry_sdk.flush() - data = span.get("data", {}) + spans = [item.payload for item in items] - assert SPANDATA.CODE_LINENO in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FILEPATH in data - assert SPANDATA.CODE_FUNCTION in data + assert len(spans) == 3 - assert type(data.get(SPANDATA.CODE_LINENO)) == int - assert data.get(SPANDATA.CODE_LINENO) > 0 - assert ( - data.get(SPANDATA.CODE_NAMESPACE) == "tests.integrations.asyncpg.test_asyncpg" - ) - assert data.get(SPANDATA.CODE_FILEPATH).endswith( - "tests/integrations/asyncpg/test_asyncpg.py" - ) + connect_span = spans[0] + insert_span = spans[1] + segment = spans[2] - is_relative_path = data.get(SPANDATA.CODE_FILEPATH)[0] != os.sep - assert is_relative_path + assert segment["name"] == "test_segment" + assert insert_span["name"].startswith("INSERT INTO") + assert connect_span["name"] == "connect" - assert ( - data.get(SPANDATA.CODE_FUNCTION) - == "test_query_source_if_duration_over_threshold" - ) + _assert_query_source(insert_span, "test_query_source_if_duration_over_threshold") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) -async def test_span_origin(sentry_init, capture_events, capture_items, span_streaming): +async def test_span_origin(sentry_init, capture_items): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.execute("SELECT 1") - await conn.fetchrow("SELECT 2") - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - - assert len(spans) == 4 - - connect_span = spans[0] - select1_span = spans[1] - select2_span = spans[2] - segment = spans[3] - - assert segment["name"] == "test_segment" - assert connect_span["name"] == "connect" - assert select1_span["name"] == "SELECT 1" - assert select2_span["name"] == "SELECT 2" + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - assert segment["attributes"]["sentry.origin"] == "manual" - assert connect_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" - assert select1_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" - assert select2_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" - else: - events = capture_events() + await conn.execute("SELECT 1") + await conn.fetchrow("SELECT 2") + await conn.close() + sentry_sdk.flush() - with start_transaction(name="test_transaction"): - conn: Connection = await connect(PG_CONNECTION_URI) + spans = [item.payload for item in items] - await conn.execute("SELECT 1") - await conn.fetchrow("SELECT 2") - await conn.close() + assert len(spans) == 4 - (event,) = events + connect_span = spans[0] + select1_span = spans[1] + select2_span = spans[2] + segment = spans[3] - assert event["contexts"]["trace"]["origin"] == "manual" + assert segment["name"] == "test_segment" + assert connect_span["name"] == "connect" + assert select1_span["name"] == "SELECT 1" + assert select2_span["name"] == "SELECT 2" - for span in event["spans"]: - assert span["origin"] == "auto.db.asyncpg" + assert segment["attributes"]["sentry.origin"] == "manual" + assert connect_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" + assert select1_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" + assert select2_span["attributes"]["sentry.origin"] == "auto.db.asyncpg" @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_multiline_query_description_normalized( - sentry_init, capture_events, capture_items, span_streaming -): - sentry_init( - integrations=[AsyncPGIntegration()], - traces_sample_rate=1.0, - trace_lifecycle="stream" if span_streaming else "static", - ) - - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.execute( - """ - SELECT - id, - name - FROM - users - WHERE - name = 'Alice' - """ - ) - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - - assert len(spans) == 3 - - connect_span = spans[0] - select_span = spans[1] - segment = spans[2] - - assert segment["name"] == "test_segment" - assert connect_span["name"] == "connect" - assert select_span["name"] == "SELECT id, name FROM users WHERE name = 'Alice'" - else: - events = capture_events() - - with start_transaction(name="test_transaction"): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.execute( - """ - SELECT - id, - name - FROM - users - WHERE - name = 'Alice' - """ - ) - await conn.close() - - (event,) = events - - spans = [ - s - for s in event["spans"] - if s["op"] == "db" and "SELECT" in s.get("description", "") - ] - assert len(spans) == 1 - assert ( - spans[0]["description"] == "SELECT id, name FROM users WHERE name = 'Alice'" - ) - - -@pytest.mark.asyncio -async def test_before_send_transaction_sees_normalized_description( - sentry_init, capture_events + sentry_init, + capture_items, ): - def before_send_transaction(event, hint): - for span in event.get("spans", []): - desc = span.get("description", "") - if "SELECT id, name FROM users" in desc: - span["description"] = "filtered" - return event - sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, - before_send_transaction=before_send_transaction, + trace_lifecycle="stream", ) - events = capture_events() - with start_transaction(name="test_transaction"): + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): conn: Connection = await connect(PG_CONNECTION_URI) await conn.execute( """ @@ -1141,213 +850,164 @@ def before_send_transaction(event, hint): name FROM users + WHERE + name = 'Alice' """ ) await conn.close() - (event,) = events - spans = [ - s - for s in event["spans"] - if s["op"] == "db" and "filtered" in s.get("description", "") - ] + sentry_sdk.flush() + + spans = [item.payload for item in items] - assert len(spans) == 1 - assert spans[0]["description"] == "filtered" + assert len(spans) == 3 + connect_span = spans[0] + select_span = spans[1] + segment = spans[2] -def _assert_query_source(span, span_streaming, expected_function): - if span_streaming: - data = span.get("attributes", {}) - lineno_key = "code.line.number" - filepath_key = "code.file.path" - else: - data = span.get("data", {}) - lineno_key = SPANDATA.CODE_LINENO - filepath_key = SPANDATA.CODE_FILEPATH + assert segment["name"] == "test_segment" + assert connect_span["name"] == "connect" + assert select_span["name"] == "SELECT id, name FROM users WHERE name = 'Alice'" - assert lineno_key in data - assert filepath_key in data + +def _assert_query_source(span, expected_function): + data = span.get("attributes", {}) + + assert "code.line.number" in data + assert "code.file.path" in data assert SPANDATA.CODE_NAMESPACE in data assert SPANDATA.CODE_FUNCTION in data - assert type(data.get(lineno_key)) == int - assert data.get(lineno_key) > 0 + assert type(data.get("code.line.number")) == int + assert data.get("code.line.number") > 0 assert data[SPANDATA.CODE_NAMESPACE] == "tests.integrations.asyncpg.test_asyncpg" - assert data.get(filepath_key).endswith("tests/integrations/asyncpg/test_asyncpg.py") - assert data.get(filepath_key)[0] != os.sep + assert data.get("code.file.path").endswith( + "tests/integrations/asyncpg/test_asyncpg.py" + ) + assert data.get("code.file.path")[0] != os.sep assert data[SPANDATA.CODE_FUNCTION] == expected_function @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_execute( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - "Alice", - "pw", - datetime.date(1990, 12, 25), - ) - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - assert len(spans) == 3 - - connect_span = spans[0] - query_span = spans[1] - segment = spans[2] - - assert connect_span["name"] == "connect" - assert query_span["name"].startswith("INSERT INTO") - assert segment["name"] == "test_segment" - assert segment["is_segment"] is True - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.execute( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - "Alice", - "pw", - datetime.date(1990, 12, 25), - ) - await conn.close() + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) + await conn.execute( + "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", + "Alice", + "pw", + datetime.date(1990, 12, 25), + ) + await conn.close() + sentry_sdk.flush() + + spans = [item.payload for item in items] + assert len(spans) == 3 - (event,) = events - spans = event["spans"] - assert len(spans) == 2 - assert spans[0]["description"] == "connect" - assert spans[1]["description"].startswith("INSERT INTO") - query_span = spans[1] + connect_span = spans[0] + query_span = spans[1] + segment = spans[2] - _assert_query_source(query_span, span_streaming, "test_query_source_execute") + assert connect_span["name"] == "connect" + assert query_span["name"].startswith("INSERT INTO") + assert segment["name"] == "test_segment" + assert segment["is_segment"] is True + + _assert_query_source(query_span, "test_query_source_execute") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_executemany( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [("Bob", "secret_pw", datetime.date(1984, 3, 1))], - ) - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - assert len(spans) == 3 - - connect_span = spans[0] - query_span = spans[1] - segment = spans[2] - - assert connect_span["name"] == "connect" - assert query_span["name"].startswith("INSERT INTO") - assert segment["name"] == "test_segment" - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [("Bob", "secret_pw", datetime.date(1984, 3, 1))], - ) - await conn.close() + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) + await conn.executemany( + "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", + [("Bob", "secret_pw", datetime.date(1984, 3, 1))], + ) + await conn.close() + sentry_sdk.flush() - (event,) = events - spans = event["spans"] - assert len(spans) == 2 - assert spans[0]["description"] == "connect" - assert spans[1]["description"].startswith("INSERT INTO") - query_span = spans[1] + spans = [item.payload for item in items] + assert len(spans) == 3 - _assert_query_source(query_span, span_streaming, "test_query_source_executemany") + connect_span = spans[0] + query_span = spans[1] + segment = spans[2] + + assert connect_span["name"] == "connect" + assert query_span["name"].startswith("INSERT INTO") + assert segment["name"] == "test_segment" + + _assert_query_source(query_span, "test_query_source_executemany") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_query_source_prepare( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ): sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.prepare("SELECT * FROM users WHERE name = $1") - await conn.close() - sentry_sdk.flush() - - spans = [item.payload for item in items] - assert len(spans) == 3 - connect_span = spans[0] - query_span = spans[1] - segment = spans[2] - - assert connect_span["name"] == "connect" - assert query_span["name"] == "SELECT * FROM users WHERE name = $1" - assert segment["name"] == "test_segment" - - assert ( - query_span["attributes"][SPANDATA.DB_QUERY_TEXT] - == "SELECT * FROM users WHERE name = $1" - ) - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) - await conn.prepare("SELECT * FROM users WHERE name = $1") - await conn.close() + items = capture_items("span") + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) + await conn.prepare("SELECT * FROM users WHERE name = $1") + await conn.close() + sentry_sdk.flush() + + spans = [item.payload for item in items] + assert len(spans) == 3 + connect_span = spans[0] + query_span = spans[1] + segment = spans[2] + + assert connect_span["name"] == "connect" + assert query_span["name"] == "SELECT * FROM users WHERE name = $1" + assert segment["name"] == "test_segment" - (event,) = events - spans = event["spans"] - assert len(spans) == 2 - assert spans[0]["description"] == "connect" - assert spans[1]["description"] == "SELECT * FROM users WHERE name = $1" - query_span = spans[1] + assert ( + query_span["attributes"][SPANDATA.DB_QUERY_TEXT] + == "SELECT * FROM users WHERE name = $1" + ) - _assert_query_source(query_span, span_streaming, "test_query_source_prepare") + _assert_query_source(query_span, "test_query_source_prepare") @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_cursor_iteration_creates_db_cursor_iter_spans( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ) -> None: """ Regression test for https://github.com/getsentry/sentry-python/issues/6576 @@ -1364,196 +1024,115 @@ async def test_cursor_iteration_creates_db_cursor_iter_spans( sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [(f"user-{i}", "pw", datetime.date(1990, 1, 1)) for i in range(20)], - ) - - async with conn.transaction(): - async for _record in conn.cursor("SELECT * FROM users", prefetch=5): - pass - - await conn.close() + items = capture_items("span") - sentry_sdk.flush() - - cursor_iter_spans = [ - item.payload - for item in items - if item.payload.get("name") == "SELECT * FROM users" - ] - - assert len(cursor_iter_spans) == 5 - for span in cursor_iter_spans: - assert span["attributes"]["sentry.op"] == OP.DB_CURSOR_ITERATOR - assert span["attributes"][SPANDATA.DB_QUERY_TEXT] == "SELECT * FROM users" - else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - conn: Connection = await connect(PG_CONNECTION_URI) + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [(f"user-{i}", "pw", datetime.date(1990, 1, 1)) for i in range(20)], - ) + await conn.executemany( + "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", + [(f"user-{i}", "pw", datetime.date(1990, 1, 1)) for i in range(20)], + ) - async with conn.transaction(): - async for _record in conn.cursor("SELECT * FROM users", prefetch=5): - pass + async with conn.transaction(): + async for _record in conn.cursor("SELECT * FROM users", prefetch=5): + pass - await conn.close() + await conn.close() - (event,) = events + sentry_sdk.flush() - cursor_iter_spans = [ - s for s in event["spans"] if s.get("description") == "SELECT * FROM users" - ] + cursor_iter_spans = [ + item.payload + for item in items + if item.payload.get("name") == "SELECT * FROM users" + ] - assert len(cursor_iter_spans) == 5 - for span in cursor_iter_spans: - assert span["op"] == OP.DB_CURSOR_ITERATOR + assert len(cursor_iter_spans) == 5 + for span in cursor_iter_spans: + assert span["attributes"]["sentry.op"] == OP.DB_CURSOR_ITERATOR + assert span["attributes"][SPANDATA.DB_QUERY_TEXT] == "SELECT * FROM users" @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_cursor_fetch_methods_create_spans( - sentry_init, capture_events, capture_items, span_streaming + sentry_init, + capture_items, ) -> None: sentry_init( integrations=[AsyncPGIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items() - - with sentry_sdk.traces.start_span(name="test_segment"): - conn: Connection = await connect(PG_CONNECTION_URI) - - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [ - ("Bob", "secret_pw", datetime.date(1984, 3, 1)), - ("Alice", "pw", datetime.date(1990, 12, 25)), - ], - ) - - async with conn.transaction(): - cur = await conn.cursor( - "SELECT * FROM users WHERE dob > $1", datetime.date(1970, 1, 1) - ) - # These exercise the `_exec` patch - await cur.fetchrow() - await cur.fetchrow() - - await conn.close() - - sentry_sdk.flush() - - spans = [item.payload for item in items] + items = capture_items() - assert len(spans) == 7 - - connect_span = spans[0] - executemany_span = spans[1] - begin_span = spans[2] - fetchrow_span_1 = spans[3] - fetchrow_span_2 = spans[4] - commit_span = spans[5] - _segment_span = spans[6] + with sentry_sdk.traces.start_span(name="test_segment"): + conn: Connection = await connect(PG_CONNECTION_URI) - assert connect_span["name"] == "connect" - assert ( - executemany_span["name"] - == "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)" - ) - assert begin_span["name"] == "BEGIN;" - assert fetchrow_span_1["name"] == "SELECT * FROM users WHERE dob > $1" - assert fetchrow_span_2["name"] == "SELECT * FROM users WHERE dob > $1" - assert commit_span["name"] == "COMMIT;" - - assert ( - fetchrow_span_1["attributes"][SPANDATA.DB_QUERY_TEXT] - == "SELECT * FROM users WHERE dob > $1" - ) - assert ( - fetchrow_span_2["attributes"][SPANDATA.DB_QUERY_TEXT] - == "SELECT * FROM users WHERE dob > $1" + await conn.executemany( + "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", + [ + ("Bob", "secret_pw", datetime.date(1984, 3, 1)), + ("Alice", "pw", datetime.date(1990, 12, 25)), + ], ) - for span in (fetchrow_span_1, fetchrow_span_2): - assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "postgresql" - assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "asyncpg" - assert span["attributes"]["sentry.op"] == OP.DB_CURSOR_FETCH - assert span["attributes"]["sentry.origin"] == "auto.db.asyncpg" - - else: - events = capture_events() + async with conn.transaction(): + cur = await conn.cursor( + "SELECT * FROM users WHERE dob > $1", datetime.date(1970, 1, 1) + ) + # These exercise the `_exec` patch + await cur.fetchrow() + await cur.fetchrow() - with start_transaction(name="test_transaction"): - conn: Connection = await connect(PG_CONNECTION_URI) + await conn.close() - await conn.executemany( - "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)", - [ - ("Bob", "secret_pw", datetime.date(1984, 3, 1)), - ("Alice", "pw", datetime.date(1990, 12, 25)), - ], - ) + sentry_sdk.flush() - async with conn.transaction(): - cur = await conn.cursor( - "SELECT * FROM users WHERE dob > $1", datetime.date(1970, 1, 1) - ) - # These exercise the `_exec` patch - await cur.fetchrow() - await cur.fetchrow() + spans = [item.payload for item in items] - await conn.close() + assert len(spans) == 7 - (event,) = events + connect_span = spans[0] + executemany_span = spans[1] + begin_span = spans[2] + fetchrow_span_1 = spans[3] + fetchrow_span_2 = spans[4] + commit_span = spans[5] + _segment_span = spans[6] - assert len(event["spans"]) == 6 + assert connect_span["name"] == "connect" + assert ( + executemany_span["name"] + == "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)" + ) + assert begin_span["name"] == "BEGIN;" + assert fetchrow_span_1["name"] == "SELECT * FROM users WHERE dob > $1" + assert fetchrow_span_2["name"] == "SELECT * FROM users WHERE dob > $1" + assert commit_span["name"] == "COMMIT;" - connect_span = event["spans"][0] - executemany_span = event["spans"][1] - begin_span = event["spans"][2] - fetchrow_span_1 = event["spans"][3] - fetchrow_span_2 = event["spans"][4] - commit_span = event["spans"][5] + assert ( + fetchrow_span_1["attributes"][SPANDATA.DB_QUERY_TEXT] + == "SELECT * FROM users WHERE dob > $1" + ) + assert ( + fetchrow_span_2["attributes"][SPANDATA.DB_QUERY_TEXT] + == "SELECT * FROM users WHERE dob > $1" + ) - assert connect_span["description"] == "connect" - assert ( - executemany_span["description"] - == "INSERT INTO users(name, password, dob) VALUES($1, $2, $3)" - ) - assert begin_span["description"] == "BEGIN;" - assert fetchrow_span_1["description"] == "SELECT * FROM users WHERE dob > $1" - assert fetchrow_span_2["description"] == "SELECT * FROM users WHERE dob > $1" - assert commit_span["description"] == "COMMIT;" - - for span in (fetchrow_span_1, fetchrow_span_2): - assert span["data"]["db.cursor"] is not None - assert span["data"]["db.system"] == "postgresql" - assert span["data"]["db.driver.name"] == "asyncpg" - assert span["op"] == OP.DB_CURSOR_FETCH - assert span["origin"] == "auto.db.asyncpg" + for span in (fetchrow_span_1, fetchrow_span_2): + assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "postgresql" + assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "asyncpg" + assert span["attributes"]["sentry.op"] == OP.DB_CURSOR_FETCH + assert span["attributes"]["sentry.origin"] == "auto.db.asyncpg" _assert_query_source( span, - span_streaming, "test_cursor_fetch_methods_create_spans", ) diff --git a/tests/integrations/django/test_basic.py b/tests/integrations/django/test_basic.py index 61c25da205..224fd1f4a5 100644 --- a/tests/integrations/django/test_basic.py +++ b/tests/integrations/django/test_basic.py @@ -34,7 +34,7 @@ from sentry_sdk.integrations.django.signals_handlers import _get_receiver_name from sentry_sdk.integrations.executing import ExecutingIntegration from sentry_sdk.profiler.utils import get_frame_name -from sentry_sdk.tracing import Span +from sentry_sdk.traces import StreamedSpan from tests.conftest import unpack_werkzeug_response from tests.integrations.django.myapp.signals import myapp_custom_signal_silenced from tests.integrations.django.myapp.wsgi import application @@ -871,7 +871,9 @@ def get_connection_params(self): return {} try: - _set_db_data(Span(), DummyBackend()) + _set_db_data( + StreamedSpan(name="name", scope=sentry_sdk.Scope()), DummyBackend() + ) except TypeError: pytest.fail("A TypeError was raised") diff --git a/tests/integrations/sqlalchemy/test_sqlalchemy.py b/tests/integrations/sqlalchemy/test_sqlalchemy.py index e08bb0a423..a8e809c71f 100644 --- a/tests/integrations/sqlalchemy/test_sqlalchemy.py +++ b/tests/integrations/sqlalchemy/test_sqlalchemy.py @@ -9,24 +9,19 @@ from sqlalchemy.orm import relationship, sessionmaker import sentry_sdk -from sentry_sdk import capture_message, start_transaction +from sentry_sdk import capture_message from sentry_sdk.consts import SPANDATA from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration -from sentry_sdk.serializer import MAX_EVENT_BYTES from sentry_sdk.tracing_utils import record_sql_queries -from sentry_sdk.utils import json_dumps -@pytest.mark.parametrize("span_streaming", [True, False]) def test_orm_queries( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, @@ -61,14 +56,9 @@ class Address(Base): assert session.query(Person).first() == bob - if span_streaming: - items = capture_items("event") - capture_message("hi") - (event,) = (item.payload for item in items) - else: - events = capture_events() - capture_message("hi") - (event,) = events + items = capture_items("event") + capture_message("hi") + (event,) = (item.payload for item in items) for crumb in event["breadcrumbs"]["values"]: del crumb["timestamp"] @@ -91,17 +81,14 @@ class Address(Base): ] -@pytest.mark.parametrize("span_streaming", [True, False]) def test_transactions( sentry_init, - capture_events, capture_items, render_span_tree, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, @@ -132,37 +119,36 @@ class Address(Base): Session = sessionmaker(bind=engine) # noqa: N806 session = Session() - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - with session.begin_nested(): - session.query(Person).first() - - for _ in range(2): - with pytest.raises(IntegrityError): - with session.begin_nested(): - session.add(Person(id=1, name="bob")) - session.add(Person(id=1, name="bob")) - - with session.begin_nested(): - session.query(Person).first() - - sentry_sdk.flush() - spans = [item.payload for item in items] - sqlalchemy_spans = [ - span - for span in spans - if span["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" - ] - for span in sqlalchemy_spans: - assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "sqlite" - assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" - assert span["attributes"][SPANDATA.DB_NAMESPACE] == ":memory:" - assert SPANDATA.SERVER_PORT not in span["attributes"] - - assert ( - render_span_tree(spans) - == """\ + items = capture_items("span") + with sentry_sdk.traces.start_span(name="custom parent"): + with session.begin_nested(): + session.query(Person).first() + + for _ in range(2): + with pytest.raises(IntegrityError): + with session.begin_nested(): + session.add(Person(id=1, name="bob")) + session.add(Person(id=1, name="bob")) + + with session.begin_nested(): + session.query(Person).first() + + sentry_sdk.flush() + spans = [item.payload for item in items] + sqlalchemy_spans = [ + span + for span in spans + if span["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" + ] + for span in sqlalchemy_spans: + assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "sqlite" + assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" + assert span["attributes"][SPANDATA.DB_NAMESPACE] == ":memory:" + assert SPANDATA.SERVER_PORT not in span["attributes"] + + assert ( + render_span_tree(spans) + == """\ - sentry.op=null: name="custom parent" - sentry.op="db": name="SAVEPOINT sa_savepoint_1" - sentry.op="db": name="SELECT person.id AS person_id, person.name AS person_name \\nFROM person\\n LIMIT ? OFFSET ?" @@ -177,61 +163,16 @@ class Address(Base): - sentry.op="db": name="SELECT person.id AS person_id, person.name AS person_name \\nFROM person\\n LIMIT ? OFFSET ?" - sentry.op="db": name="RELEASE SAVEPOINT sa_savepoint_4"\ """ - ) - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - with session.begin_nested(): - session.query(Person).first() - - for _ in range(2): - with pytest.raises(IntegrityError): - with session.begin_nested(): - session.add(Person(id=1, name="bob")) - session.add(Person(id=1, name="bob")) - - with session.begin_nested(): - session.query(Person).first() - - (event,) = events - - for span in event["spans"]: - assert span["data"][SPANDATA.DB_SYSTEM] == "sqlite" - assert span["data"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" - assert span["data"][SPANDATA.DB_NAME] == ":memory:" - assert SPANDATA.SERVER_ADDRESS not in span["data"] - assert SPANDATA.SERVER_PORT not in span["data"] - - assert ( - render_span_tree(event["spans"], event["contexts"]["trace"]) - == """\ -- op=null: description=null - - op="db": description="SAVEPOINT sa_savepoint_1" - - op="db": description="SELECT person.id AS person_id, person.name AS person_name \\nFROM person\\n LIMIT ? OFFSET ?" - - op="db": description="RELEASE SAVEPOINT sa_savepoint_1" - - op="db": description="SAVEPOINT sa_savepoint_2" - - op="db": description="INSERT INTO person (id, name) VALUES (?, ?)" - - op="db": description="ROLLBACK TO SAVEPOINT sa_savepoint_2" - - op="db": description="SAVEPOINT sa_savepoint_3" - - op="db": description="INSERT INTO person (id, name) VALUES (?, ?)" - - op="db": description="ROLLBACK TO SAVEPOINT sa_savepoint_3" - - op="db": description="SAVEPOINT sa_savepoint_4" - - op="db": description="SELECT person.id AS person_id, person.name AS person_name \\nFROM person\\n LIMIT ? OFFSET ?" - - op="db": description="RELEASE SAVEPOINT sa_savepoint_4"\ -""" - ) + ) -@pytest.mark.parametrize("span_streaming", [True, False]) def test_transactions_no_engine_url( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", _experiments={ "record_sql_params": True, }, @@ -263,169 +204,65 @@ class Address(Base): Session = sessionmaker(bind=engine) # noqa: N806 session = Session() - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - with session.begin_nested(): - session.query(Person).first() - - for _ in range(2): - with pytest.raises(IntegrityError): - with session.begin_nested(): - session.add(Person(id=1, name="bob")) - session.add(Person(id=1, name="bob")) - - with session.begin_nested(): - session.query(Person).first() - - sentry_sdk.flush() - spans = [item.payload for item in items] - sqlalchemy_spans = [ - span - for span in spans - if span["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" - ] - for span in sqlalchemy_spans: - assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "sqlite" - assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" - assert SPANDATA.DB_NAME not in span["attributes"] - assert SPANDATA.SERVER_PORT not in span["attributes"] - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - with session.begin_nested(): - session.query(Person).first() - - for _ in range(2): - with pytest.raises(IntegrityError): - with session.begin_nested(): - session.add(Person(id=1, name="bob")) - session.add(Person(id=1, name="bob")) - - with session.begin_nested(): - session.query(Person).first() - - (event,) = events - for span in event["spans"]: - assert span["data"][SPANDATA.DB_SYSTEM] == "sqlite" - assert span["data"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" - assert SPANDATA.DB_NAME not in span["data"] - assert SPANDATA.SERVER_ADDRESS not in span["data"] - assert SPANDATA.SERVER_PORT not in span["data"] - - -@pytest.mark.parametrize("span_streaming", [True, False]) + items = capture_items("span") + with sentry_sdk.traces.start_span(name="custom parent"): + with session.begin_nested(): + session.query(Person).first() + + for _ in range(2): + with pytest.raises(IntegrityError): + with session.begin_nested(): + session.add(Person(id=1, name="bob")) + session.add(Person(id=1, name="bob")) + + with session.begin_nested(): + session.query(Person).first() + + sentry_sdk.flush() + spans = [item.payload for item in items] + sqlalchemy_spans = [ + span + for span in spans + if span["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" + ] + for span in sqlalchemy_spans: + assert span["attributes"][SPANDATA.DB_SYSTEM_NAME] == "sqlite" + assert span["attributes"][SPANDATA.DB_DRIVER_NAME] == "pysqlite" + assert SPANDATA.DB_NAME not in span["attributes"] + assert SPANDATA.SERVER_PORT not in span["attributes"] + + def test_long_sql_query_preserved( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( traces_sample_rate=1, integrations=[SqlalchemyIntegration()], - trace_lifecycle="stream" if span_streaming else "static", - ) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - with engine.connect() as con: - con.execute( - text(" UNION ".join("SELECT {}".format(i) for i in range(100))) - ) - - sentry_sdk.flush() - spans = [item.payload for item in items] - name = spans[0]["name"] - assert name.startswith("SELECT 0 UNION SELECT 1") - assert name.endswith("SELECT 98 UNION SELECT 99") - else: - events = capture_events() - with start_transaction(name="test"): - with engine.connect() as con: - con.execute( - text(" UNION ".join("SELECT {}".format(i) for i in range(100))) - ) - - (event,) = events - description = event["spans"][0]["description"] - assert description.startswith("SELECT 0 UNION SELECT 1") - assert description.endswith("SELECT 98 UNION SELECT 99") - - -@pytest.mark.parametrize("max_value_length", [1024, None]) -def test_large_event_not_truncated(sentry_init, capture_events, max_value_length): - sentry_init( - traces_sample_rate=1, - integrations=[SqlalchemyIntegration()], - max_value_length=max_value_length, + trace_lifecycle="stream", ) - events = capture_events() - - long_str = "x" * (1034) - - scope = sentry_sdk.get_isolation_scope() - - @scope.add_event_processor - def processor(event, hint): - event["message"] = long_str - return event engine = create_engine( "sqlite:///:memory:", connect_args={"check_same_thread": False} ) - with start_transaction(name="test"): + items = capture_items("span") + with sentry_sdk.traces.start_span(name="custom parent"): with engine.connect() as con: - for _ in range(1500): - con.execute( - text(" UNION ".join("SELECT {}".format(i) for i in range(100))) - ) - - (event,) = events - - assert len(json_dumps(event)) > MAX_EVENT_BYTES - - # Some spans are discarded. - assert len(event["spans"]) == 1000 - - # Span descriptions are not truncated. - description = event["spans"][0]["description"] - assert len(description) == 1583 - assert description.startswith("SELECT 0") - assert description.endswith("SELECT 98 UNION SELECT 99") - - description = event["spans"][999]["description"] - assert len(description) == 1583 - assert description.startswith("SELECT 0") - assert description.endswith("SELECT 98 UNION SELECT 99") - - if max_value_length: - # Smoke check that truncation of other fields has not changed. - assert len(event["message"]) == 1024 - - # The _meta for other truncated fields should be there as well. - assert event["_meta"]["message"] == { - "": { - "len": 1034, - "rem": [["!limit", "x", 1021, 1024]], - } - } - else: - assert len(event["message"]) == 1034 + con.execute(text(" UNION ".join("SELECT {}".format(i) for i in range(100)))) + + sentry_sdk.flush() + spans = [item.payload for item in items] + name = spans[0]["name"] + assert name.startswith("SELECT 0 UNION SELECT 1") + assert name.endswith("SELECT 98 UNION SELECT 99") -@pytest.mark.parametrize("span_streaming", [True, False]) def test_engine_name_not_string( sentry_init, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) engine = create_engine( @@ -437,118 +274,71 @@ def test_engine_name_not_string( con.execute(text("SELECT 0")) -@pytest.mark.parametrize("span_streaming", [True, False]) def test_query_source_disabled( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_options = { "integrations": [SqlalchemyIntegration()], "traces_sample_rate": 1.0, "enable_db_query_source": False, "db_query_source_threshold_ms": 0, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } sentry_init(**sentry_options) - if span_streaming: - items = capture_items("span") + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() - bob = Person(name="Bob") - session.add(bob) + bob = Person(name="Bob") + session.add(bob) - assert session.query(Person).first() == bob + assert session.query(Person).first() == bob - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER not in attributes - assert SPANDATA.CODE_NAMESPACE not in attributes - assert SPANDATA.CODE_FILE_PATH not in attributes - assert SPANDATA.CODE_FUNCTION not in attributes - break - else: - raise AssertionError("No db span found") + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] + assert SPANDATA.CODE_LINE_NUMBER not in attributes + assert SPANDATA.CODE_NAMESPACE not in attributes + assert SPANDATA.CODE_FILE_PATH not in attributes + assert SPANDATA.CODE_FUNCTION not in attributes + break else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - assert session.query(Person).first() == bob - - (event,) = events - - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) - - assert SPANDATA.CODE_LINENO not in data - assert SPANDATA.CODE_NAMESPACE not in data - assert SPANDATA.CODE_FILEPATH not in data - assert SPANDATA.CODE_FUNCTION not in data - break - else: - raise AssertionError("No db span found") + raise AssertionError("No db span found") @pytest.mark.parametrize("enable_db_query_source", [None, True]) -@pytest.mark.parametrize("span_streaming", [True, False]) def test_query_source_enabled( sentry_init, - capture_events, capture_items, enable_db_query_source, - span_streaming, ): sentry_options = { "integrations": [SqlalchemyIntegration()], "traces_sample_rate": 1.0, "db_query_source_threshold_ms": 0, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } if enable_db_query_source is not None: @@ -556,216 +346,115 @@ def test_query_source_enabled( sentry_init(**sentry_options) - if span_streaming: - items = capture_items("span") + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - bob = Person(name="Bob") - session.add(bob) - - assert session.query(Person).first() == bob - - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER in attributes - assert SPANDATA.CODE_NAMESPACE in attributes - assert SPANDATA.CODE_FILE_PATH in attributes - assert SPANDATA.CODE_FUNCTION in attributes - break - else: - raise AssertionError("No db span found") - else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) - bob = Person(name="Bob") - session.add(bob) + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() - assert session.query(Person).first() == bob + bob = Person(name="Bob") + session.add(bob) - (event,) = events + assert session.query(Person).first() == bob - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] - assert SPANDATA.CODE_LINENO in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FILEPATH in data - assert SPANDATA.CODE_FUNCTION in data - break - else: - raise AssertionError("No db span found") + assert SPANDATA.CODE_LINE_NUMBER in attributes + assert SPANDATA.CODE_NAMESPACE in attributes + assert SPANDATA.CODE_FILE_PATH in attributes + assert SPANDATA.CODE_FUNCTION in attributes + break + else: + raise AssertionError("No db span found") -@pytest.mark.parametrize("span_streaming", [True, False]) def test_query_source( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) + + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() + + bob = Person(name="Bob") + session.add(bob) + + assert session.query(Person).first() == bob + + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] + + assert SPANDATA.CODE_LINE_NUMBER in attributes + assert SPANDATA.CODE_NAMESPACE in attributes + assert SPANDATA.CODE_FILE_PATH in attributes + assert SPANDATA.CODE_FUNCTION in attributes + + assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int + assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 + assert ( + attributes.get(SPANDATA.CODE_NAMESPACE) + == "tests.integrations.sqlalchemy.test_sqlalchemy" ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - assert session.query(Person).first() == bob - - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER in attributes - assert SPANDATA.CODE_NAMESPACE in attributes - assert SPANDATA.CODE_FILE_PATH in attributes - assert SPANDATA.CODE_FUNCTION in attributes - - assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int - assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 - assert ( - attributes.get(SPANDATA.CODE_NAMESPACE) - == "tests.integrations.sqlalchemy.test_sqlalchemy" - ) - assert attributes.get(SPANDATA.CODE_FILE_PATH).endswith( - "tests/integrations/sqlalchemy/test_sqlalchemy.py" - ) - - is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep - assert is_relative_path - - assert attributes.get(SPANDATA.CODE_FUNCTION) == "test_query_source" - break - else: - raise AssertionError("No db span found") - else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} + assert attributes.get(SPANDATA.CODE_FILE_PATH).endswith( + "tests/integrations/sqlalchemy/test_sqlalchemy.py" ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - assert session.query(Person).first() == bob - - (event,) = events - - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) - - assert SPANDATA.CODE_LINENO in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FILEPATH in data - assert SPANDATA.CODE_FUNCTION in data - assert type(data.get(SPANDATA.CODE_LINENO)) == int - assert data.get(SPANDATA.CODE_LINENO) > 0 - assert ( - data.get(SPANDATA.CODE_NAMESPACE) - == "tests.integrations.sqlalchemy.test_sqlalchemy" - ) - assert data.get(SPANDATA.CODE_FILEPATH).endswith( - "tests/integrations/sqlalchemy/test_sqlalchemy.py" - ) + is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep + assert is_relative_path - is_relative_path = data.get(SPANDATA.CODE_FILEPATH)[0] != os.sep - assert is_relative_path - - assert data.get(SPANDATA.CODE_FUNCTION) == "test_query_source" - break - else: - raise AssertionError("No db span found") + assert attributes.get(SPANDATA.CODE_FUNCTION) == "test_query_source" + break + else: + raise AssertionError("No db span found") -@pytest.mark.parametrize("span_streaming", [True, False]) def test_query_source_with_module_in_search_path( sentry_init, - capture_events, capture_items, - span_streaming, ): """ Test that query source is relative to the path of the module it ran in @@ -775,7 +464,7 @@ def test_query_source_with_module_in_search_path( traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) from sqlalchemy_helpers.helpers import ( @@ -783,449 +472,241 @@ def test_query_source_with_module_in_search_path( query_first_model_from_session, ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - - add_model_to_session(bob, session) - - assert query_first_model_from_session(Person, session) == bob - - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER in attributes - assert SPANDATA.CODE_NAMESPACE in attributes - assert SPANDATA.CODE_FILE_PATH in attributes - assert SPANDATA.CODE_FUNCTION in attributes - - assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int - assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 - assert ( - attributes.get(SPANDATA.CODE_NAMESPACE) - == "sqlalchemy_helpers.helpers" - ) - assert ( - attributes.get(SPANDATA.CODE_FILE_PATH) - == "sqlalchemy_helpers/helpers.py" - ) - - is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep - assert is_relative_path - - assert ( - attributes.get(SPANDATA.CODE_FUNCTION) - == "query_first_model_from_session" - ) - break - else: - raise AssertionError("No db span found") - else: - events = capture_events() - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 + items = capture_items("span") + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) - bob = Person(name="Bob") + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() - add_model_to_session(bob, session) + bob = Person(name="Bob") - assert query_first_model_from_session(Person, session) == bob + add_model_to_session(bob, session) - (event,) = events + assert query_first_model_from_session(Person, session) == bob - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] - assert SPANDATA.CODE_LINENO in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FILEPATH in data - assert SPANDATA.CODE_FUNCTION in data + assert SPANDATA.CODE_LINE_NUMBER in attributes + assert SPANDATA.CODE_NAMESPACE in attributes + assert SPANDATA.CODE_FILE_PATH in attributes + assert SPANDATA.CODE_FUNCTION in attributes - assert type(data.get(SPANDATA.CODE_LINENO)) == int - assert data.get(SPANDATA.CODE_LINENO) > 0 - assert data.get(SPANDATA.CODE_NAMESPACE) == "sqlalchemy_helpers.helpers" - assert ( - data.get(SPANDATA.CODE_FILEPATH) == "sqlalchemy_helpers/helpers.py" - ) + assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int + assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 + assert ( + attributes.get(SPANDATA.CODE_NAMESPACE) == "sqlalchemy_helpers.helpers" + ) + assert ( + attributes.get(SPANDATA.CODE_FILE_PATH) + == "sqlalchemy_helpers/helpers.py" + ) - is_relative_path = data.get(SPANDATA.CODE_FILEPATH)[0] != os.sep - assert is_relative_path + is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep + assert is_relative_path - assert ( - data.get(SPANDATA.CODE_FUNCTION) == "query_first_model_from_session" - ) - break - else: - raise AssertionError("No db span found") + assert ( + attributes.get(SPANDATA.CODE_FUNCTION) + == "query_first_model_from_session" + ) + break + else: + raise AssertionError("No db span found") -@pytest.mark.parametrize("span_streaming", [True, False]) def test_no_query_source_if_duration_too_short( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=100, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 + items = capture_items("span") - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - class fake_record_sql_queries: # noqa: N801 - def __init__(self, *args, **kwargs): - self._ctx_mgr = record_sql_queries(*args, **kwargs) - - def __enter__(self): - self.span = self._ctx_mgr.__enter__() - self.span._start_timestamp = datetime(2024, 1, 1, microsecond=0) - self.span._end_timestamp = datetime(2024, 1, 1, microsecond=99999) - return self.span - - def __exit__(self, type, value, traceback): - self.span._end_timestamp = None - self._ctx_mgr.__exit__(type, value, traceback) - - with mock.patch( - "sentry_sdk.integrations.sqlalchemy.record_sql_queries", - fake_record_sql_queries, - ): - assert session.query(Person).first() == bob - - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER not in attributes - assert SPANDATA.CODE_NAMESPACE not in attributes - assert SPANDATA.CODE_FILE_PATH not in attributes - assert SPANDATA.CODE_FUNCTION not in attributes - break - else: - raise AssertionError("No db span found") + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - else: - events = capture_events() - - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 - - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) - - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) - class fake_record_sql_queries: # noqa: N801 - def __init__(self, *args, **kwargs): - self._ctx_mgr = record_sql_queries(*args, **kwargs) + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() - def __enter__(self): - self.span = self._ctx_mgr.__enter__() - self.span.start_timestamp = datetime(2024, 1, 1, microsecond=0) - return self.span + bob = Person(name="Bob") + session.add(bob) - def __exit__(self, type, value, traceback): - self._ctx_mgr.__exit__(type, value, traceback) - self.span.timestamp = datetime(2024, 1, 1, microsecond=99999) + class fake_record_sql_queries: # noqa: N801 + def __init__(self, *args, **kwargs): + self._ctx_mgr = record_sql_queries(*args, **kwargs) - with mock.patch( - "sentry_sdk.integrations.sqlalchemy.record_sql_queries", - fake_record_sql_queries, - ): - assert session.query(Person).first() == bob + def __enter__(self): + self.span = self._ctx_mgr.__enter__() + self.span._start_timestamp = datetime(2024, 1, 1, microsecond=0) + self.span._end_timestamp = datetime(2024, 1, 1, microsecond=99999) + return self.span - (event,) = events + def __exit__(self, type, value, traceback): + self.span._end_timestamp = None + self._ctx_mgr.__exit__(type, value, traceback) - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) + with mock.patch( + "sentry_sdk.integrations.sqlalchemy.record_sql_queries", + fake_record_sql_queries, + ): + assert session.query(Person).first() == bob - assert SPANDATA.CODE_LINENO not in data - assert SPANDATA.CODE_NAMESPACE not in data - assert SPANDATA.CODE_FILEPATH not in data - assert SPANDATA.CODE_FUNCTION not in data - break - else: - raise AssertionError("No db span found") + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] + + assert SPANDATA.CODE_LINE_NUMBER not in attributes + assert SPANDATA.CODE_NAMESPACE not in attributes + assert SPANDATA.CODE_FILE_PATH not in attributes + assert SPANDATA.CODE_FUNCTION not in attributes + break + else: + raise AssertionError("No db span found") -@pytest.mark.parametrize("span_streaming", [True, False]) def test_query_source_if_duration_over_threshold( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], traces_sample_rate=1.0, enable_db_query_source=True, db_query_source_threshold_ms=100, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) - if span_streaming: - items = capture_items("span") + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - Base = declarative_base() # noqa: N806 + with sentry_sdk.traces.start_span(name="custom parent"): + Base = declarative_base() # noqa: N806 - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) + class Person(Base): + __tablename__ = "person" + id = Column(Integer, primary_key=True) + name = Column(String(250), nullable=False) - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} - ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - class fake_record_sql_queries: # noqa: N801 - def __init__(self, *args, **kwargs): - self._ctx_mgr = record_sql_queries(*args, **kwargs) - - def __enter__(self): - self.span = self._ctx_mgr.__enter__() - self.span._start_timestamp = datetime(2024, 1, 1, microsecond=0) - self.span._end_timestamp = datetime(2024, 1, 1, microsecond=101000) - return self.span - - def __exit__(self, type, value, traceback): - self.span._end_timestamp = None - self._ctx_mgr.__exit__(type, value, traceback) - - with mock.patch( - "sentry_sdk.integrations.sqlalchemy.record_sql_queries", - fake_record_sql_queries, - ): - assert session.query(Person).first() == bob - - sentry_sdk.flush() - spans = [item.payload for item in items] - for span in spans: - if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( - "SELECT person" - ): - attributes = span["attributes"] - - assert SPANDATA.CODE_LINE_NUMBER in attributes - assert SPANDATA.CODE_NAMESPACE in attributes - assert SPANDATA.CODE_FILE_PATH in attributes - assert SPANDATA.CODE_FUNCTION in attributes - - assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int - assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 - assert ( - attributes.get(SPANDATA.CODE_NAMESPACE) - == "tests.integrations.sqlalchemy.test_sqlalchemy" - ) - assert attributes.get(SPANDATA.CODE_FILE_PATH).endswith( - "tests/integrations/sqlalchemy/test_sqlalchemy.py" - ) - - is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep - assert is_relative_path - - assert ( - attributes.get(SPANDATA.CODE_FUNCTION) - == "test_query_source_if_duration_over_threshold" - ) - break - else: - raise AssertionError("No db span found") - else: - events = capture_events() + engine = create_engine( + "sqlite:///:memory:", connect_args={"check_same_thread": False} + ) + Base.metadata.create_all(engine) + + Session = sessionmaker(bind=engine) # noqa: N806 + session = Session() + + bob = Person(name="Bob") + session.add(bob) - with start_transaction(name="test_transaction", sampled=True): - Base = declarative_base() # noqa: N806 + class fake_record_sql_queries: # noqa: N801 + def __init__(self, *args, **kwargs): + self._ctx_mgr = record_sql_queries(*args, **kwargs) - class Person(Base): - __tablename__ = "person" - id = Column(Integer, primary_key=True) - name = Column(String(250), nullable=False) + def __enter__(self): + self.span = self._ctx_mgr.__enter__() + self.span._start_timestamp = datetime(2024, 1, 1, microsecond=0) + self.span._end_timestamp = datetime(2024, 1, 1, microsecond=101000) + return self.span - engine = create_engine( - "sqlite:///:memory:", connect_args={"check_same_thread": False} + def __exit__(self, type, value, traceback): + self.span._end_timestamp = None + self._ctx_mgr.__exit__(type, value, traceback) + + with mock.patch( + "sentry_sdk.integrations.sqlalchemy.record_sql_queries", + fake_record_sql_queries, + ): + assert session.query(Person).first() == bob + + sentry_sdk.flush() + spans = [item.payload for item in items] + for span in spans: + if span["attributes"].get("sentry.op") == "db" and span["name"].startswith( + "SELECT person" + ): + attributes = span["attributes"] + + assert SPANDATA.CODE_LINE_NUMBER in attributes + assert SPANDATA.CODE_NAMESPACE in attributes + assert SPANDATA.CODE_FILE_PATH in attributes + assert SPANDATA.CODE_FUNCTION in attributes + + assert type(attributes.get(SPANDATA.CODE_LINE_NUMBER)) == int + assert attributes.get(SPANDATA.CODE_LINE_NUMBER) > 0 + assert ( + attributes.get(SPANDATA.CODE_NAMESPACE) + == "tests.integrations.sqlalchemy.test_sqlalchemy" + ) + assert attributes.get(SPANDATA.CODE_FILE_PATH).endswith( + "tests/integrations/sqlalchemy/test_sqlalchemy.py" + ) + + is_relative_path = attributes.get(SPANDATA.CODE_FILE_PATH)[0] != os.sep + assert is_relative_path + + assert ( + attributes.get(SPANDATA.CODE_FUNCTION) + == "test_query_source_if_duration_over_threshold" ) - Base.metadata.create_all(engine) - - Session = sessionmaker(bind=engine) # noqa: N806 - session = Session() - - bob = Person(name="Bob") - session.add(bob) - - class fake_record_sql_queries: # noqa: N801 - def __init__(self, *args, **kwargs): - self._ctx_mgr = record_sql_queries(*args, **kwargs) - - def __enter__(self): - self.span = self._ctx_mgr.__enter__() - self.span.start_timestamp = datetime(2024, 1, 1, microsecond=0) - return self.span - - def __exit__(self, type, value, traceback): - self._ctx_mgr.__exit__(type, value, traceback) - self.span.timestamp = datetime(2024, 1, 1, microsecond=101000) - - with mock.patch( - "sentry_sdk.integrations.sqlalchemy.record_sql_queries", - fake_record_sql_queries, - ): - assert session.query(Person).first() == bob - - (event,) = events - - for span in event["spans"]: - if span.get("op") == "db" and span.get("description").startswith( - "SELECT person" - ): - data = span.get("data", {}) - - assert SPANDATA.CODE_LINENO in data - assert SPANDATA.CODE_NAMESPACE in data - assert SPANDATA.CODE_FILEPATH in data - assert SPANDATA.CODE_FUNCTION in data - - assert type(data.get(SPANDATA.CODE_LINENO)) == int - assert data.get(SPANDATA.CODE_LINENO) > 0 - assert ( - data.get(SPANDATA.CODE_NAMESPACE) - == "tests.integrations.sqlalchemy.test_sqlalchemy" - ) - assert data.get(SPANDATA.CODE_FILEPATH).endswith( - "tests/integrations/sqlalchemy/test_sqlalchemy.py" - ) - - is_relative_path = data.get(SPANDATA.CODE_FILEPATH)[0] != os.sep - assert is_relative_path - - assert ( - data.get(SPANDATA.CODE_FUNCTION) - == "test_query_source_if_duration_over_threshold" - ) - break - else: - raise AssertionError("No db span found") - - -@pytest.mark.parametrize("span_streaming", [True, False]) + break + else: + raise AssertionError("No db span found") + + def test_span_origin( sentry_init, - capture_events, capture_items, - span_streaming, ): sentry_init( integrations=[SqlalchemyIntegration()], traces_sample_rate=1.0, - trace_lifecycle="stream" if span_streaming else "static", + trace_lifecycle="stream", ) engine = create_engine( "sqlite:///:memory:", connect_args={"check_same_thread": False} ) - if span_streaming: - items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - with engine.connect() as con: - con.execute(text("SELECT 0")) - - sentry_sdk.flush() - spans = [item.payload for item in items] - - assert spans[0]["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" - assert spans[1]["attributes"]["sentry.origin"] == "manual" - else: - events = capture_events() - with start_transaction(name="foo"): - with engine.connect() as con: - con.execute(text("SELECT 0")) + items = capture_items("span") + with sentry_sdk.traces.start_span(name="custom parent"): + with engine.connect() as con: + con.execute(text("SELECT 0")) - (event,) = events + sentry_sdk.flush() + spans = [item.payload for item in items] - assert event["contexts"]["trace"]["origin"] == "manual" - assert event["spans"][0]["origin"] == "auto.db.sqlalchemy" + assert spans[0]["attributes"]["sentry.origin"] == "auto.db.sqlalchemy" + assert spans[1]["attributes"]["sentry.origin"] == "manual"