fix: Handle mypy 2.0 related failures

sentry_sdk/_compat.py

@@ -54,7 +54,7 @@ def enabled(option: str) -> bool:
             except Exception:
                 pass
 
-        return value and str(value).lower() not in FALSE_VALUES
+        return value and str(value).lower() not in FALSE_VALUES  # type: ignore[return-value]
 
     # When `threads` is passed in as a uwsgi option,
     # `enable-threads` is implied on.

sentry_sdk/client.py

@@ -864,7 +864,7 @@ def capture_event(
 
         :returns: An event ID. May be `None` if there is no DSN set or of if the SDK decided to discard the event for other reasons. In such situations setting `debug=True` on `init()` may help.
         """
-        hint: "Hint" = dict(hint or ())
+        hint = dict(hint or ())
 
         if not self._should_capture(event, hint, scope):
             return None
@@ -948,7 +948,7 @@ def _capture_telemetry(
         if ty == "log":
             before_send = get_before_send_log(self.options)
         elif ty == "metric":
-            before_send = get_before_send_metric(self.options)  # type: ignore
+            before_send = get_before_send_metric(self.options)
 
         if before_send is not None:
             telemetry = before_send(telemetry, {})  # type: ignore
@@ -960,9 +960,9 @@ def _capture_telemetry(
         if ty == "log":
             batcher = self.log_batcher
         elif ty == "metric":
-            batcher = self.metrics_batcher  # type: ignore
+            batcher = self.metrics_batcher
         elif ty == "span":
-            batcher = self.span_batcher  # type: ignore
+            batcher = self.span_batcher
 
         if batcher is not None:
             batcher.add(telemetry)  # type: ignore

sentry_sdk/crons/api.py

@@ -18,7 +18,7 @@ def _create_check_in_event(
     monitor_config: "Optional[MonitorConfig]" = None,
 ) -> "Event":
     options = sentry_sdk.get_client().options
-    check_in_id: str = check_in_id or uuid.uuid4().hex
+    check_in_id = check_in_id or uuid.uuid4().hex
 
     check_in: "Event" = {
         "type": "check_in",

sentry_sdk/hub.py

@@ -146,7 +146,7 @@ def __init__(
                 scope = get_isolation_scope().fork()
                 current_scope = get_current_scope().fork()
         else:
-            client = client_or_hub  # type: ignore
+            client = client_or_hub
             get_global_scope().set_client(client)
 
         if scope is None:  # so there is no Hub cloning going on

sentry_sdk/integrations/_wsgi_common.py

@@ -9,8 +9,11 @@
 
 try:
     from django.http.request import RawPostDataException
+
+    _RAW_DATA_EXCEPTIONS = (RawPostDataException, ValueError)
 except ImportError:
     RawPostDataException = None
+    _RAW_DATA_EXCEPTIONS = (ValueError,)
 
 from typing import TYPE_CHECKING
 
@@ -110,7 +113,7 @@ def extract_into_event(self, event: "Event") -> None:
             raw_data = None
             try:
                 raw_data = self.raw_data()
-            except (RawPostDataException, ValueError):
+            except _RAW_DATA_EXCEPTIONS:
                 # If DjangoRestFramework is used it already read the body for us
                 # so reading it here will fail. We can ignore this.
                 pass
@@ -175,7 +178,7 @@ def json(self) -> "Optional[Any]":
 
             try:
                 raw_data = self.raw_data()
-            except (RawPostDataException, ValueError):
+            except _RAW_DATA_EXCEPTIONS:
                 # The body might have already been read, in which case this will
                 # fail
                 raw_data = None

sentry_sdk/integrations/cloud_resource_context.py

@@ -176,6 +176,7 @@ def _get_gcp_context(cls) -> "Dict[str, str]":
             "cloud.platform": CLOUD_PLATFORM.GCP_COMPUTE_ENGINE,
         }
 
+        gcp_metadata = cls.gcp_metadata
         try:
             if cls.gcp_metadata is None:
                 r = cls.http.request(
@@ -187,30 +188,29 @@ def _get_gcp_context(cls) -> "Dict[str, str]":
                 if r.status != 200:
                     return ctx
 
-                cls.gcp_metadata = json.loads(r.data.decode("utf-8"))
+                gcp_metadata = json.loads(r.data.decode("utf-8"))
+                cls.gcp_metadata = gcp_metadata
 
             try:
-                ctx["cloud.account.id"] = cls.gcp_metadata["project"]["projectId"]
+                ctx["cloud.account.id"] = gcp_metadata["project"]["projectId"]
             except Exception:
                 pass
 
             try:
-                ctx["cloud.availability_zone"] = cls.gcp_metadata["instance"][
-                    "zone"
-                ].split("/")[-1]
+                ctx["cloud.availability_zone"] = gcp_metadata["instance"]["zone"].split(
+                    "/"
+                )[-1]
             except Exception:
                 pass
 
             try:
                 # only populated in google cloud run
-                ctx["cloud.region"] = cls.gcp_metadata["instance"]["region"].split("/")[
-                    -1
-                ]
+                ctx["cloud.region"] = gcp_metadata["instance"]["region"].split("/")[-1]
             except Exception:
                 pass
 
             try:
-                ctx["host.id"] = cls.gcp_metadata["instance"]["id"]
+                ctx["host.id"] = gcp_metadata["instance"]["id"]
             except Exception:
                 pass
 

sentry_sdk/integrations/django/__init__.py

@@ -120,9 +120,9 @@ class DjangoIntegration(Integration):
     origin_db = f"auto.db.{identifier}"
 
     transaction_style = ""
-    middleware_spans = None
-    signals_spans = None
-    cache_spans = None
+    middleware_spans: "Optional[bool]" = None
+    signals_spans: "Optional[bool]" = None
+    cache_spans: "Optional[bool]" = None
     signals_denylist: "list[signals.Signal]" = []
 
     def __init__(

sentry_sdk/integrations/django/asgi.py

@@ -44,7 +44,7 @@
     iscoroutinefunction = inspect.iscoroutinefunction
     markcoroutinefunction = inspect.markcoroutinefunction
 else:
-    iscoroutinefunction = asyncio.iscoroutinefunction  # type: ignore[assignment]
+    iscoroutinefunction = asyncio.iscoroutinefunction
 
     def markcoroutinefunction(func: "_F") -> "_F":
         func._is_coroutine = asyncio.coroutines._is_coroutine  # type: ignore

sentry_sdk/integrations/dramatiq.py

@@ -70,7 +70,7 @@ def sentry_patched_broker__init__(
             # RedisBroker does not.
             if len(args) == 1:
                 middleware = args[0]
-                args = []  # type: ignore
+                args = ()
             else:
                 middleware = None
 

sentry_sdk/integrations/gnu_backtrace.py

@@ -16,14 +16,14 @@
 FUNCTION_RE = r"[^@]+?"
 HEX_ADDRESS = r"\s+@\s+0x[0-9a-fA-F]+"
 
-FRAME_RE = r"""
+_FRAME_RE_PATTERN = r"""
 ^(?P<index>\d+)\.\s+(?P<function>{FUNCTION_RE}){HEX_ADDRESS}(?:\s+in\s+(?P<package>.+))?$
 """.format(
     FUNCTION_RE=FUNCTION_RE,
     HEX_ADDRESS=HEX_ADDRESS,
 )
 
-FRAME_RE = re.compile(FRAME_RE, re.MULTILINE | re.VERBOSE)
+FRAME_RE = re.compile(_FRAME_RE_PATTERN, re.MULTILINE | re.VERBOSE)
 
 
 class GnuBacktraceIntegration(Integration):

sentry_sdk/integrations/grpc/__init__.py

@@ -152,7 +152,7 @@ def patched_aio_server(  # type: ignore
         **kwargs: "P.kwargs",
     ) -> "Server":
         server_interceptor = AsyncServerInterceptor()
-        interceptors: "Sequence[grpc.ServerInterceptor]" = [
+        interceptors = [
             server_interceptor,
             *(interceptors or []),
         ]

sentry_sdk/integrations/mcp.py

@@ -495,7 +495,7 @@ async def _handler_wrapper(
                         uri = original_kwargs.get("uri")
 
                     protocol = None
-                    if hasattr(uri, "scheme"):
+                    if uri is not None and hasattr(uri, "scheme"):
                         protocol = uri.scheme
                     elif handler_name and "://" in handler_name:
                         protocol = handler_name.split("://")[0]
@@ -638,7 +638,7 @@ def _patch_fastmcp() -> None:
     This function patches the _get_prompt_mcp and _read_resource_mcp methods
     to add instrumentation for those handlers.
     """
-    if hasattr(FastMCP, "_get_prompt_mcp"):
+    if FastMCP is not None and hasattr(FastMCP, "_get_prompt_mcp"):
         original_get_prompt_mcp = FastMCP._get_prompt_mcp
 
         @wraps(original_get_prompt_mcp)
@@ -655,7 +655,7 @@ async def patched_get_prompt_mcp(
 
         FastMCP._get_prompt_mcp = patched_get_prompt_mcp
 
-    if hasattr(FastMCP, "_read_resource_mcp"):
+    if FastMCP is not None and hasattr(FastMCP, "_read_resource_mcp"):
         original_read_resource_mcp = FastMCP._read_resource_mcp
 
         @wraps(original_read_resource_mcp)

sentry_sdk/integrations/openai_agents/__init__.py

@@ -99,72 +99,80 @@ def setup_once() -> None:
             0,
             8,
         ):
-
-            @wraps(run_loop.get_all_tools)
-            async def new_wrapped_get_all_tools(
-                agent: "agents.Agent",
-                context_wrapper: "agents.RunContextWrapper",
-            ) -> "list[agents.Tool]":
-                return await _get_all_tools(
-                    run_loop.get_all_tools, agent, context_wrapper
+            if run_loop is not None:
+
+                @wraps(run_loop.get_all_tools)
+                async def new_wrapped_get_all_tools(
+                    agent: "agents.Agent",
+                    context_wrapper: "agents.RunContextWrapper",
+                ) -> "list[agents.Tool]":
+                    return await _get_all_tools(
+                        run_loop.get_all_tools, agent, context_wrapper
+                    )
+
+                agents.run.get_all_tools = new_wrapped_get_all_tools
+
+                @wraps(run_loop.run_single_turn)
+                async def new_wrapped_run_single_turn(
+                    *args: "Any", **kwargs: "Any"
+                ) -> "SingleStepResult":
+                    return await _run_single_turn(
+                        run_loop.run_single_turn, *args, **kwargs
+                    )
+
+                agents.run.run_single_turn = new_wrapped_run_single_turn
+
+                @wraps(run_loop.run_single_turn_streamed)
+                async def new_wrapped_run_single_turn_streamed(
+                    *args: "Any", **kwargs: "Any"
+                ) -> "SingleStepResult":
+                    return await _run_single_turn_streamed(
+                        run_loop.run_single_turn_streamed, *args, **kwargs
+                    )
+
+                agents.run.run_single_turn_streamed = (
+                    new_wrapped_run_single_turn_streamed
                 )
 
-            agents.run.get_all_tools = new_wrapped_get_all_tools
-
-            @wraps(turn_preparation.get_model)
-            def new_wrapped_get_model(
-                agent: "agents.Agent", run_config: "agents.RunConfig"
-            ) -> "agents.Model":
-                return _get_model(turn_preparation.get_model, agent, run_config)
-
-            agents.run_internal.run_loop.get_model = new_wrapped_get_model
-
-            @wraps(run_loop.run_single_turn)
-            async def new_wrapped_run_single_turn(
-                *args: "Any", **kwargs: "Any"
-            ) -> "SingleStepResult":
-                return await _run_single_turn(run_loop.run_single_turn, *args, **kwargs)
+            if turn_preparation is not None:
 
-            agents.run.run_single_turn = new_wrapped_run_single_turn
+                @wraps(turn_preparation.get_model)
+                def new_wrapped_get_model(
+                    agent: "agents.Agent", run_config: "agents.RunConfig"
+                ) -> "agents.Model":
+                    return _get_model(turn_preparation.get_model, agent, run_config)
 
-            @wraps(run_loop.run_single_turn_streamed)
-            async def new_wrapped_run_single_turn_streamed(
-                *args: "Any", **kwargs: "Any"
-            ) -> "SingleStepResult":
-                return await _run_single_turn_streamed(
-                    run_loop.run_single_turn_streamed, *args, **kwargs
-                )
+                agents.run_internal.run_loop.get_model = new_wrapped_get_model
 
-            agents.run.run_single_turn_streamed = new_wrapped_run_single_turn_streamed
+            if turn_resolution is not None:
+                original_execute_handoffs = turn_resolution.execute_handoffs
 
-            original_execute_handoffs = turn_resolution.execute_handoffs
+                @wraps(original_execute_handoffs)
+                async def new_wrapped_execute_handoffs(
+                    *args: "Any", **kwargs: "Any"
+                ) -> "SingleStepResult":
+                    return await _execute_handoffs(
+                        original_execute_handoffs, *args, **kwargs
+                    )
 
-            @wraps(original_execute_handoffs)
-            async def new_wrapped_execute_handoffs(
-                *args: "Any", **kwargs: "Any"
-            ) -> "SingleStepResult":
-                return await _execute_handoffs(
-                    original_execute_handoffs, *args, **kwargs
+                agents.run_internal.turn_resolution.execute_handoffs = (
+                    new_wrapped_execute_handoffs
                 )
 
-            agents.run_internal.turn_resolution.execute_handoffs = (
-                new_wrapped_execute_handoffs
-            )
+                original_execute_final_output = turn_resolution.execute_final_output
 
-            original_execute_final_output = turn_resolution.execute_final_output
+                @wraps(turn_resolution.execute_final_output)
+                async def new_wrapped_final_output(
+                    *args: "Any", **kwargs: "Any"
+                ) -> "SingleStepResult":
+                    return await _execute_final_output(
+                        original_execute_final_output, *args, **kwargs
+                    )
 
-            @wraps(turn_resolution.execute_final_output)
-            async def new_wrapped_final_output(
-                *args: "Any", **kwargs: "Any"
-            ) -> "SingleStepResult":
-                return await _execute_final_output(
-                    original_execute_final_output, *args, **kwargs
+                agents.run_internal.turn_resolution.execute_final_output = (
+                    new_wrapped_final_output
                 )
 
-            agents.run_internal.turn_resolution.execute_final_output = (
-                new_wrapped_final_output
-            )
-
             return
 
         original_get_all_tools = AgentRunner._get_all_tools

sentry_sdk/integrations/opentelemetry/integration.py

@@ -52,4 +52,6 @@ def _setup_sentry_tracing() -> None:
 
 def _setup_instrumentors() -> None:
     for instrumentor, kwargs in CONFIGURABLE_INSTRUMENTATIONS.items():
+        if instrumentor is None:
+            continue
         instrumentor().instrument(**kwargs)

sentry_sdk/integrations/sanic.py

@@ -58,7 +58,7 @@
 class SanicIntegration(Integration):
     identifier = "sanic"
     origin = f"auto.http.{identifier}"
-    version = None
+    version: "Optional[tuple[int, ...]]" = None
 
     def __init__(
         self, unsampled_statuses: "Optional[Container[int]]" = frozenset({404})

sentry_sdk/integrations/starlette.py

@@ -475,7 +475,7 @@ def _is_async_callable(obj: "Any") -> bool:
         obj = obj.func
 
     return iscoroutinefunction(obj) or (
-        callable(obj) and iscoroutinefunction(obj.__call__)
+        callable(obj) and iscoroutinefunction(obj.__call__)  # type: ignore[operator]
     )