merge master

Author: alexander-alderman-webb

sentry_sdk/consts.py

@@ -1180,6 +1180,8 @@ class OP:
     COHERE_CHAT_COMPLETIONS_CREATE = "ai.chat_completions.create.cohere"
     COHERE_EMBEDDINGS_CREATE = "ai.embeddings.create.cohere"
     DB = "db"
+    DB_CURSOR_ITERATOR = "db.cursor.iter"
+    DB_CURSOR_FETCH = "db.cursor.fetch"
     DB_REDIS = "db.redis"
     EVENT_DJANGO = "event.django"
     FUNCTION = "function"

sentry_sdk/integrations/asyncpg.py

@@ -20,8 +20,12 @@
 )
 
 try:
-    import asyncpg  # type: ignore[import-not-found]
-    from asyncpg.cursor import BaseCursor  # type: ignore
+    import asyncpg  # type: ignore
+    from asyncpg.cursor import (  # type: ignore
+        BaseCursor,
+        Cursor,
+        CursorIterator,
+    )
 
 except ImportError:
     raise DidNotEnable("asyncpg not installed.")
@@ -169,6 +173,13 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
             return await f(*args, **kwargs)
 
         cursor = args[0]
+        if type(cursor) is CursorIterator:
+            span_op_override_value = OP.DB_CURSOR_ITERATOR
+        elif type(cursor) is Cursor:
+            span_op_override_value = OP.DB_CURSOR_FETCH
+        else:
+            span_op_override_value = None
+
         query = _normalize_query(cursor._query)
         with record_sql_queries(
             cursor=cursor,
@@ -178,6 +189,7 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
             executemany=False,
             record_cursor_repr=True,
             span_origin=AsyncPGIntegration.origin,
+            span_op_override_value=span_op_override_value,
         ) as span:
             _set_db_data(span, cursor._connection)
             res = await f(*args, **kwargs)

sentry_sdk/integrations/langchain.py

@@ -210,6 +210,8 @@ class LangchainIntegration(Integration):
     identifier = "langchain"
     origin = f"auto.ai.{identifier}"
 
+    _ignored_exceptions: "set[type[Exception]]" = set()
+
     def __init__(
         self: "LangchainIntegration",
         include_prompts: bool = True,
@@ -262,17 +264,22 @@ def gc_span_map(self) -> None:
                 self._exit_span(span, run_id)
 
     def _handle_error(self, run_id: "UUID", error: "Any") -> None:
+        is_ignored = isinstance(error, tuple(LangchainIntegration._ignored_exceptions))
+
         with capture_internal_exceptions():
             if not run_id or run_id not in self.span_map:
                 return
 
             span = self.span_map[run_id]
 
-            sentry_sdk.capture_exception(
-                error, span._scope if isinstance(span, StreamedSpan) else span.scope
-            )
+            if is_ignored:
+                span.__exit__(None, None, None)
+            else:
+                sentry_sdk.capture_exception(
+                    error, span._scope if isinstance(span, StreamedSpan) else span.scope
+                )
+                span.__exit__(type(error), error, error.__traceback__)
 
-            span.__exit__(type(error), error, error.__traceback__)
             del self.span_map[run_id]
 
     def _normalize_langchain_message(self, message: "BaseMessage") -> "Any":

sentry_sdk/integrations/langgraph.py

@@ -10,6 +10,9 @@
 )
 from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration
+
+# This is fine because langgraph depends on langchain-base, and LangchainIntegration only imports from langchain-base.
+from sentry_sdk.integrations.langchain import LangchainIntegration
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.traces import StreamedSpan
 from sentry_sdk.tracing_utils import (
@@ -19,6 +22,7 @@
 from sentry_sdk.utils import safe_serialize
 
 try:
+    from langgraph.errors import GraphBubbleUp
     from langgraph.graph import StateGraph
     from langgraph.pregel import Pregel
 except ImportError:
@@ -34,6 +38,7 @@ def __init__(self: "LanggraphIntegration", include_prompts: bool = True) -> None
 
     @staticmethod
     def setup_once() -> None:
+        LangchainIntegration._ignored_exceptions.add(GraphBubbleUp)
         # LangGraph lets users create agents using a StateGraph or the Functional API.
         # StateGraphs are then compiled to a CompiledStateGraph. Both CompiledStateGraph and
         # the functional API execute on a Pregel instance. Pregel is the runtime for the graph

sentry_sdk/tracing_utils.py

@@ -134,6 +134,7 @@ def record_sql_queries(
     executemany: bool,
     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]":
     # TODO: Bring back capturing of params by default
     client = sentry_sdk.get_client()
@@ -167,13 +168,15 @@ def record_sql_queries(
             name="<unknown SQL query>" if query is None else query,
             attributes={
                 "sentry.origin": span_origin,
-                "sentry.op": OP.DB,
+                "sentry.op": span_op_override_value
+                if span_op_override_value
+                else OP.DB,
             },
         ) as span:
             yield span
     else:
         with sentry_sdk.start_span(
-            op=OP.DB,
+            op=span_op_override_value if span_op_override_value is not None else OP.DB,
             name=query,
             origin=span_origin,
         ) as span:

tests/integrations/asyncpg/test_asyncpg.py

@@ -21,7 +21,7 @@
 
 import sentry_sdk
 from sentry_sdk import capture_message, start_transaction
-from sentry_sdk.consts import SPANDATA
+from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations.asyncpg import AsyncPGIntegration
 from sentry_sdk.tracing_utils import record_sql_queries
 from tests.conftest import ApproxDict
@@ -1361,75 +1361,57 @@ async def test_query_source_prepare(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("span_streaming", [True, False])
-async def test_cursor__bind_exec_creates_spans(
+async def test_cursor_iteration_creates_db_cursor_iter_spans(
     sentry_init, capture_events, capture_items, span_streaming
 ) -> None:
     """
-    Exercises the bind_exec patch through the iterator that's created in asyncpg when "for record in conn.cursor" is called.
-    See https://github.com/MagicStack/asyncpg/blob/db8ecc2a38e16fb0c090aef6f5506547c2831c24/asyncpg/cursor.py#L234
+    Regression test for https://github.com/getsentry/sentry-python/issues/6576
+
+    When iterating a server-side cursor with a small prefetch, asyncpg fetches
+    rows in batches. Each batch triggers BaseCursor._bind_exec (on first query) and
+    BaseCursor._exec (second query onwards) through CursorIterator.__anext__, which creates a
+    span with the same query description. The resulting burst of identical spans
+    causes Sentry's N+1 query detector to raise a false positive.
+
+    To mitigate, we set the "op"/"sentry.op" to `db.cursor.iter` instead of `db`
+    so that the sentry backend can exclude these spans from n+1 detection.
     """
     sentry_init(
         integrations=[AsyncPGIntegration()],
         traces_sample_rate=1.0,
-        enable_db_query_source=True,
-        db_query_source_threshold_ms=0,
         _experiments={
             "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.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)),
-                ],
+                [(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 WHERE dob > $1",
-                    datetime.date(1970, 1, 1),
-                ):
+                async for _record in conn.cursor("SELECT * FROM users", prefetch=5):
                     pass
 
             await conn.close()
-        sentry_sdk.flush()
 
-        spans = [item.payload for item in items]
-
-        assert len(spans) == 6
-
-        connect_span = spans[0]
-        executemany_span = spans[1]
-        begin_span = spans[2]
-        bind_exec_span = spans[3]
-        commit_span = spans[4]
-        segment = spans[5]
+        sentry_sdk.flush()
 
-        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 bind_exec_span["name"] == "SELECT * FROM users WHERE dob > $1"
-        assert commit_span["name"] == "COMMIT;"
-        assert segment["name"] == "test_segment"
+        cursor_iter_spans = [
+            item.payload
+            for item in items
+            if item.payload.get("name") == "SELECT * FROM users"
+        ]
 
-        assert bind_exec_span["attributes"]["sentry.origin"] == "auto.db.asyncpg"
-        assert bind_exec_span["attributes"]["sentry.op"] == "db"
-        assert bind_exec_span["attributes"]["db.system.name"] == "postgresql"
-        assert bind_exec_span["attributes"]["db.driver.name"] == "asyncpg"
-        assert bind_exec_span["attributes"]["server.address"] == PG_HOST
-        assert bind_exec_span["attributes"]["server.port"] == PG_PORT
-        assert bind_exec_span["attributes"]["db.namespace"] == PG_NAME
-        assert bind_exec_span["attributes"]["db.user"] == PG_USER
+        assert len(cursor_iter_spans) == 5
+        for span in cursor_iter_spans:
+            assert span["attributes"]["sentry.op"] == OP.DB_CURSOR_ITERATOR
     else:
         events = capture_events()
 
@@ -1438,57 +1420,28 @@ async def test_cursor__bind_exec_creates_spans(
 
             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)),
-                ],
+                [(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 WHERE dob > $1",
-                    datetime.date(1970, 1, 1),
-                ):
+                async for _record in conn.cursor("SELECT * FROM users", prefetch=5):
                     pass
 
             await conn.close()
 
         (event,) = events
 
-        assert len(event["spans"]) == 5
-
-        connect_span = event["spans"][0]
-        executemany_span = event["spans"][1]
-        begin_span = event["spans"][2]
-        bind_exec_span = event["spans"][3]
-        commit_span = event["spans"][4]
+        cursor_iter_spans = [
+            s for s in event["spans"] if s.get("description") == "SELECT * FROM users"
+        ]
 
-        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 bind_exec_span["description"] == "SELECT * FROM users WHERE dob > $1"
-        assert commit_span["description"] == "COMMIT;"
-
-        assert bind_exec_span["origin"] == "auto.db.asyncpg"
-        assert bind_exec_span["data"]["db.system"] == "postgresql"
-        assert bind_exec_span["data"]["db.driver.name"] == "asyncpg"
-        assert bind_exec_span["data"]["server.address"] == PG_HOST
-        assert bind_exec_span["data"]["server.port"] == PG_PORT
-        assert bind_exec_span["data"]["db.name"] == PG_NAME
-        assert bind_exec_span["data"]["db.user"] == PG_USER
-
-    _assert_query_source(
-        bind_exec_span,
-        span_streaming,
-        "test_cursor__bind_exec_creates_spans",
-    )
+        assert len(cursor_iter_spans) == 5
+        for span in cursor_iter_spans:
+            assert span["op"] == OP.DB_CURSOR_ITERATOR
 
 
 @pytest.mark.asyncio
-async def test_cursor__exec_methods_create_spans(sentry_init, capture_events) -> None:
+async def test_cursor_fetch_methods_create_spans(sentry_init, capture_events) -> None:
     sentry_init(
         integrations=[AsyncPGIntegration()],
         traces_sample_rate=1.0,
@@ -1543,9 +1496,10 @@ async def test_cursor__exec_methods_create_spans(sentry_init, capture_events) ->
         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"
         _assert_query_source(
             span,
             False,
-            "test_cursor__exec_methods_create_spans",
+            "test_cursor_fetch_methods_create_spans",
         )

tests/integrations/langgraph/test_langgraph.py

@@ -3,6 +3,10 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import HumanMessage
+from langchain_core.outputs import ChatResult
+from langgraph.errors import GraphBubbleUp
 
 import sentry_sdk
 from sentry_sdk import start_transaction
@@ -125,6 +129,15 @@ async def ainvoke(self, state, config=None):
         return {"messages": [MockMessage("Async Pregel response")]}
 
 
+class InterruptingChatModel(BaseChatModel):
+    @property
+    def _llm_type(self) -> str:
+        return "interrupting-chat-model"
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
+        raise GraphBubbleUp("interrupt")
+
+
 def test_langgraph_integration_init():
     """Test LanggraphIntegration initialization with different parameters."""
     integration = LanggraphIntegration()
@@ -2104,3 +2117,17 @@ def original_invoke(self, *args, **kwargs):
     assert len(parsed_messages) == 1
     assert "small message 5" in str(parsed_messages[0])
     assert tx["_meta"]["spans"]["0"]["data"]["gen_ai.request.messages"][""]["len"] == 5
+
+
+def test_graph_bubble_up_ignored(sentry_init, capture_items):
+    sentry_init(
+        integrations=[LanggraphIntegration()],
+    )
+
+    events = capture_items("event")
+
+    model = InterruptingChatModel()
+    with pytest.raises(GraphBubbleUp):
+        model.invoke([HumanMessage(content="hi")])
+
+    assert len(events) == 0