Skip to content

feat(sqlalchemy): Support span streaming#6132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexander-alderman-webb merged 25 commits into from
Apr 29, 2026
Merged

feat(sqlalchemy): Support span streaming #6132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
a139626
feat(sqlalchemy): Support span streaming
alexander-alderman-webb Apr 24, 2026
38b5933
.
alexander-alderman-webb Apr 24, 2026
3d89b9e
.
alexander-alderman-webb Apr 24, 2026
b1fa3b5
.
alexander-alderman-webb Apr 24, 2026
9929f55
.
alexander-alderman-webb Apr 24, 2026
75c0cc1
.
alexander-alderman-webb Apr 24, 2026
ad4e14d
fix mypy
alexander-alderman-webb Apr 24, 2026
2bf7c77
.
alexander-alderman-webb Apr 24, 2026
f873d09
.
alexander-alderman-webb Apr 24, 2026
56ee084
add type ignores
alexander-alderman-webb Apr 24, 2026
2c22c16
move query source before exit
alexander-alderman-webb Apr 24, 2026
f9faa2f
.
alexander-alderman-webb Apr 24, 2026
064dd83
use separate path for streaming
alexander-alderman-webb Apr 24, 2026
4b965c4
use separate path for streaming
alexander-alderman-webb Apr 24, 2026
98c67f0
remove print
alexander-alderman-webb Apr 24, 2026
2877f7b
Merge branch 'master' into webb/sqlalchemy/span-first
alexander-alderman-webb Apr 27, 2026
f20dfc6
update tests
alexander-alderman-webb Apr 27, 2026
6322ad0
add missing code.namespace assertions
alexander-alderman-webb Apr 27, 2026
193f790
use non-deprecated attributes
alexander-alderman-webb Apr 28, 2026
4a7414e
use old attributes for legacy path
alexander-alderman-webb Apr 28, 2026
5efa060
.
alexander-alderman-webb Apr 28, 2026
639318f
consistently early exit
alexander-alderman-webb Apr 28, 2026
e71988a
use non-deprecated attributes in tests
alexander-alderman-webb Apr 28, 2026
ff174f8
stop setting undocumented attributes
alexander-alderman-webb Apr 28, 2026
47f5a6b
add comment
alexander-alderman-webb Apr 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 43 additions & 15 deletions sentry_sdk/integrations/sqlalchemy.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,16 @@
from sentry_sdk.consts import SPANSTATUS, SPANDATA
from sentry_sdk.integrations import _check_minimum_version, Integration, DidNotEnable
from sentry_sdk.tracing_utils import add_query_source, record_sql_queries
from sentry_sdk.tracing_utils import (
add_query_source,
record_sql_queries_supporting_streaming,
)
from sentry_sdk.utils import (
capture_internal_exceptions,
ensure_integration_enabled,
parse_version,
)
from sentry_sdk.traces import StreamedSpan, SpanStatus
from sentry_sdk.tracing import Span

try:
from sqlalchemy.engine import Engine # type: ignore
Expand All@@ -20,8 +25,7 @@
from typing import Any
from typing import ContextManager
from typing import Optional

from sentry_sdk.tracing import Span
from typing import Union


class SqlalchemyIntegration(Integration):
Expand All@@ -48,7 +52,7 @@ def _before_cursor_execute(
executemany: bool,
*args: "Any",
) -> None:
ctx_mgr = record_sql_queries(
ctx_mgr = record_sql_queries_supporting_streaming(
cursor,
statement,
parameters,
Expand DownExpand Up@@ -78,12 +82,19 @@ def _after_cursor_execute(
context, "_sentry_sql_span_manager", None
)

# 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):
with capture_internal_exceptions():
add_query_source(span)

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

span: "Optional[Span]" = getattr(context, "_sentry_sql_span", None)
if span is not None:
if isinstance(span, Span):
with capture_internal_exceptions():
add_query_source(span)

Expand All@@ -96,7 +107,10 @@ def _handle_error(context: "Any", *args: "Any") -> None:
span: "Optional[Span]" = getattr(execution_context, "_sentry_sql_span", None)

if span is not None:
span.set_status(SPANSTATUS.INTERNAL_ERROR)
if isinstance(span, StreamedSpan):
span.status = SpanStatus.ERROR
else:
span.set_status(SPANSTATUS.INTERNAL_ERROR)
Comment thread
alexander-alderman-webb marked this conversation as resolved.

# _after_cursor_execute does not get called for crashing SQL stmts. Judging
# from SQLAlchemy codebase it does seem like any error coming into this
Comment thread
alexander-alderman-webb marked this conversation as resolved.
Expand DownExpand Up@@ -132,29 +146,43 @@ def _get_db_system(name: str) -> "Optional[str]":
return None


def _set_db_data(span: "Span", conn: "Any") -> None:
def _set_db_data(span: "Union[Span, StreamedSpan]", conn: "Any") -> None:
db_system = _get_db_system(conn.engine.name)
if db_system is not None:
span.set_data(SPANDATA.DB_SYSTEM, db_system)

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

try:
driver = conn.dialect.driver
if driver:
span.set_data(SPANDATA.DB_DRIVER_NAME, driver)
set_on_span(SPANDATA.DB_DRIVER_NAME, driver)
except Exception:
pass

if conn.engine.url is None:
return

db_name = conn.engine.url.database
if db_name is not None:
span.set_data(SPANDATA.DB_NAME, db_name)
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)

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

server_port = conn.engine.url.port
if server_port is not None:
span.set_data(SPANDATA.SERVER_PORT, server_port)
set_on_span(SPANDATA.SERVER_PORT, server_port)
78 changes: 75 additions & 3 deletions sentry_sdk/tracing_utils.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -170,6 +170,65 @@
yield span


# Mirrors record_sql_queries() temporarily so the Django and asyncpg integrations don't crash with span streaming enabled.
# Once both are ported, remove record_sql_queries() and rename record_sql_queries_supporting_streaming() to record_sql_queries().
@contextlib.contextmanager
def record_sql_queries_supporting_streaming(
Comment thread
alexander-alderman-webb marked this conversation as resolved.
cursor: "Any",
query: "Any",
params_list: "Any",
paramstyle: "Optional[str]",
executemany: bool,
record_cursor_repr: bool = False,
span_origin: str = "manual",
) -> "Generator[Union[sentry_sdk.tracing.Span, sentry_sdk.traces.StreamedSpan], None, None]":
# TODO: Bring back capturing of params by default
client = sentry_sdk.get_client()
if client.options["_experiments"].get("record_sql_params", False):
if not params_list or params_list == [None]:
params_list = None

if paramstyle == "pyformat":
paramstyle = "format"
else:
params_list = None
paramstyle = None

query = _format_sql(cursor, query)

data = {}
if params_list is not None:
data["db.params"] = params_list
if paramstyle is not None:
data["db.paramstyle"] = paramstyle
if executemany:
data["db.executemany"] = True
if record_cursor_repr and cursor is not None:
data["db.cursor"] = cursor
Comment thread
alexander-alderman-webb marked this conversation as resolved.

with capture_internal_exceptions():
sentry_sdk.add_breadcrumb(message=query, category="query", data=data)

if has_span_streaming_enabled(client.options):
with sentry_sdk.traces.start_span(
name="<unknown SQL query>" if query is None else query,
attributes={
"sentry.origin": span_origin,
"sentry.op": OP.DB,
},
) as span:
yield span

Check warning on line 220 in sentry_sdk/tracing_utils.py

View check run for this annotation

@sentry/warden/ warden: find-bugs

Streaming SQL span drops db.params, db.paramstyle, db.executemany, and db.cursor attributes 

In the streaming branch, the `data` dict (db.params, db.paramstyle, db.executemany, db.cursor) is built but never attached to the span — only `sentry.origin` and `sentry.op` are set as attributes. The legacy branch (else) correctly applies them via `span.set_data(k, v)`. As a consequence, when `trace_lifecycle == 'stream'` is enabled, SQL spans lose all parameter/cursor/executemany diagnostic information that users rely on for debugging, while the same instrumentation works on the non-streaming path. This is a behavioral regression that silently degrades observability on the streaming code path.

Check warning on line 220 in sentry_sdk/tracing_utils.py

View check run for this annotation

@sentry/warden/ warden: code-review

Streaming branch drops db.params/db.paramstyle/db.executemany/db.cursor span data 

In the `has_span_streaming_enabled` branch, the `data` dict is built (lines 198-207) and used for the breadcrumb, but is never attached to the span — only `sentry.origin` and `sentry.op` are passed as attributes. The non-streaming branch attaches all of `data` via `span.set_data(k, v)`. As a result, when span streaming is enabled, SQL spans will lose `db.params`, `db.paramstyle`, `db.executemany`, and `db.cursor` attributes, which is a behavioral/backwards-compatibility regression in telemetry data.
Comment thread
alexander-alderman-webb marked this conversation as resolved.
else:
with sentry_sdk.start_span(
op=OP.DB,
name=query,
Comment thread
alexander-alderman-webb marked this conversation as resolved.
origin=span_origin,
) as span:
for k, v in data.items():
span.set_data(k, v)
yield span


def maybe_create_breadcrumbs_from_span(
scope: "sentry_sdk.Scope", span: "sentry_sdk.tracing.Span"
) -> None:
Expand DownExpand Up@@ -316,22 +375,35 @@
span.set_attribute(SPANDATA.CODE_FUNCTION, frame.f_code.co_name)


def add_query_source(span: "sentry_sdk.tracing.Span") -> None:
def add_query_source(
span: "Union[sentry_sdk.tracing.Span, sentry_sdk.traces.StreamedSpan]",
) -> None:
"""
Adds OTel compatible source code information to a database query span
"""
client = sentry_sdk.get_client()
if not client.is_active():
return

if span.timestamp is None or span.start_timestamp is None:
if isinstance(span, LegacySpan):
# 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 LegacySpan case,
# it should already be finished.
if span.timestamp is None:
return

if span.start_timestamp is None:
return

should_add_query_source = client.options.get("enable_db_query_source", True)
if not should_add_query_source:
return

duration = span.timestamp - span.start_timestamp
end_timestamp = (
datetime.now(timezone.utc) if span.timestamp is None else span.timestamp
)

duration = end_timestamp - span.start_timestamp
threshold = client.options.get("db_query_source_threshold_ms", 0)
slow_query = duration / timedelta(milliseconds=1) > threshold

Expand Down
27 changes: 19 additions & 8 deletions tests/conftest.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -476,23 +476,34 @@ def maybe_monkeypatched_threading(request):

@pytest.fixture
def render_span_tree():
def inner(event):
assert event["type"] == "transaction"
def inner(spans, root_span=None):
streamed_spans = False
if root_span is None:
streamed_spans = True

by_parent = {}
for span in event["spans"]:
for span in spans:
if "parent_span_id" not in span:
root_span = span
continue
Comment thread
alexander-alderman-webb marked this conversation as resolved.

Comment thread
alexander-alderman-webb marked this conversation as resolved.
by_parent.setdefault(span["parent_span_id"], []).append(span)

def render_span(span):
yield "- op={}: description={}".format(
json.dumps(span.get("op")), json.dumps(span.get("description"))
)
if streamed_spans:
yield "- sentry.op={}: name={}".format(
json.dumps(span["attributes"].get("sentry.op")),
json.dumps(span["name"]),
)
else:
yield "- op={}: description={}".format(
json.dumps(span.get("op")), json.dumps(span.get("description"))
)

for subspan in by_parent.get(span["span_id"]) or ():
for line in render_span(subspan):
yield " {}".format(line)

root_span = event["contexts"]["trace"]

return "\n".join(render_span(root_span))

return inner
Expand Down
3 changes: 2 additions & 1 deletion tests/integrations/django/asgi/test_asgi.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -292,8 +292,9 @@ async def test_async_middleware_spans(

(transaction,) = events

assert transaction["type"] == "transaction"
assert (
render_span_tree(transaction)
render_span_tree(transaction["spans"], transaction["contexts"]["trace"])
== """\
- op="http.server": description=null
- op="event.django": description="django.db.reset_queries"
Expand Down
33 changes: 25 additions & 8 deletions tests/integrations/django/test_basic.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -457,7 +457,7 @@ def test_response_trace(sentry_init, client, capture_events, render_span_tree):

assert (
'- op="view.response.render": description="serialize response"'
in render_span_tree(events[0])
in render_span_tree(events[0]["spans"], events[0]["contexts"]["trace"])
)


Expand DownExpand Up@@ -596,7 +596,9 @@ def test_django_connect_trace(sentry_init, client, capture_events, render_span_t
data = span.get("data")
assert data.get(SPANDATA.DB_SYSTEM) == "postgresql"

assert '- op="db": description="connect"' in render_span_tree(event)
assert '- op="db": description="connect"' in render_span_tree(
event["spans"], event["contexts"]["trace"]
)


@pytest.mark.forked
Expand DownExpand Up@@ -954,7 +956,9 @@ def test_render_spans(sentry_init, client, capture_events, render_span_tree):
events = capture_events()
client.get(url)
transaction = events[0]
assert expected_line in render_span_tree(transaction)
assert expected_line in render_span_tree(
transaction["spans"], transaction["contexts"]["trace"]
)


@pytest.mark.skipif(DJANGO_VERSION < (1, 9), reason="Requires Django >= 1.9")
Expand DownExpand Up@@ -1034,7 +1038,10 @@ def test_middleware_spans(sentry_init, client, capture_events, render_span_tree)
message, transaction = events

assert message["message"] == "hi"
assert render_span_tree(transaction) == EXPECTED_MIDDLEWARE_SPANS
assert (
render_span_tree(transaction["spans"], transaction["contexts"]["trace"])
== EXPECTED_MIDDLEWARE_SPANS
)


def test_middleware_spans_disabled(sentry_init, client, capture_events):
Expand DownExpand Up@@ -1075,7 +1082,10 @@ def test_signals_spans(sentry_init, client, capture_events, render_span_tree):
message, transaction = events

assert message["message"] == "hi"
assert render_span_tree(transaction) == EXPECTED_SIGNALS_SPANS
assert (
render_span_tree(transaction["spans"], transaction["contexts"]["trace"])
== EXPECTED_SIGNALS_SPANS
)

assert transaction["spans"][0]["op"] == "event.django"
assert transaction["spans"][0]["description"] == "django.db.reset_queries"
Expand DownExpand Up@@ -1127,7 +1137,10 @@ def test_signals_spans_filtering(sentry_init, client, capture_events, render_spa

(transaction,) = events

assert render_span_tree(transaction) == EXPECTED_SIGNALS_SPANS_FILTERED
assert (
render_span_tree(transaction["spans"], transaction["contexts"]["trace"])
== EXPECTED_SIGNALS_SPANS_FILTERED
)

assert transaction["spans"][0]["op"] == "event.django"
assert transaction["spans"][0]["description"] == "django.db.reset_queries"
Expand DownExpand Up@@ -1206,7 +1219,9 @@ def test_custom_urlconf_middleware(
event = events.pop(0)
assert event["transaction"] == "/custom/ok"
if middleware_spans:
assert "custom_urlconf_middleware" in render_span_tree(event)
assert "custom_urlconf_middleware" in render_span_tree(
event["spans"], event["contexts"]["trace"]
)

_content, status, _headers = unpack_werkzeug_response(client.get("/custom/exc"))
assert status.lower() == "500 internal server error"
Expand All@@ -1216,7 +1231,9 @@ def test_custom_urlconf_middleware(
assert error_event["exception"]["values"][-1]["mechanism"]["type"] == "django"
assert transaction_event["transaction"] == "/custom/exc"
if middleware_spans:
assert "custom_urlconf_middleware" in render_span_tree(transaction_event)
assert "custom_urlconf_middleware" in render_span_tree(
transaction_event["spans"], transaction_event["contexts"]["trace"]
)

settings.MIDDLEWARE.pop(0)

Expand Down
Loading
Loading