Merge branch 'srothh/async-transport-integration' into srothh/async-task-identifier

Author: srothh

sentry_sdk/transport.py

@@ -590,9 +590,115 @@ def flush(
             self._worker.flush(timeout, callback)
 
 
+class HttpTransport(BaseHttpTransport):
+    if TYPE_CHECKING:
+        _pool: Union[PoolManager, ProxyManager]
+
+    def _get_pool_options(self: Self) -> Dict[str, Any]:
+
+        num_pools = self.options.get("_experiments", {}).get("transport_num_pools")
+        options = {
+            "num_pools": 2 if num_pools is None else int(num_pools),
+            "cert_reqs": "CERT_REQUIRED",
+            "timeout": urllib3.Timeout(total=self.TIMEOUT),
+        }
+
+        socket_options: Optional[List[Tuple[int, int, int | bytes]]] = None
+
+        if self.options["socket_options"] is not None:
+            socket_options = self.options["socket_options"]
+
+        if self.options["keep_alive"]:
+            if socket_options is None:
+                socket_options = []
+
+            used_options = {(o[0], o[1]) for o in socket_options}
+            for default_option in KEEP_ALIVE_SOCKET_OPTIONS:
+                if (default_option[0], default_option[1]) not in used_options:
+                    socket_options.append(default_option)
+
+        if socket_options is not None:
+            options["socket_options"] = socket_options
+
+        options["ca_certs"] = (
+            self.options["ca_certs"]  # User-provided bundle from the SDK init
+            or os.environ.get("SSL_CERT_FILE")
+            or os.environ.get("REQUESTS_CA_BUNDLE")
+            or certifi.where()
+        )
+
+        options["cert_file"] = self.options["cert_file"] or os.environ.get(
+            "CLIENT_CERT_FILE"
+        )
+        options["key_file"] = self.options["key_file"] or os.environ.get(
+            "CLIENT_KEY_FILE"
+        )
+
+        return options
+
+    def _make_pool(self: Self) -> Union[PoolManager, ProxyManager]:
+        if self.parsed_dsn is None:
+            raise ValueError("Cannot create HTTP-based transport without valid DSN")
+
+        proxy = None
+        no_proxy = self._in_no_proxy(self.parsed_dsn)
+
+        # try HTTPS first
+        https_proxy = self.options["https_proxy"]
+        if self.parsed_dsn.scheme == "https" and (https_proxy != ""):
+            proxy = https_proxy or (not no_proxy and getproxies().get("https"))
+
+        # maybe fallback to HTTP proxy
+        http_proxy = self.options["http_proxy"]
+        if not proxy and (http_proxy != ""):
+            proxy = http_proxy or (not no_proxy and getproxies().get("http"))
+
+        opts = self._get_pool_options()
+
+        if proxy:
+            proxy_headers = self.options["proxy_headers"]
+            if proxy_headers:
+                opts["proxy_headers"] = proxy_headers
+
+            if proxy.startswith("socks"):
+                use_socks_proxy = True
+                try:
+                    # Check if PySocks dependency is available
+                    from urllib3.contrib.socks import SOCKSProxyManager
+                except ImportError:
+                    use_socks_proxy = False
+                    logger.warning(
+                        "You have configured a SOCKS proxy (%s) but support for SOCKS proxies is not installed. Disabling proxy support. Please add `PySocks` (or `urllib3` with the `[socks]` extra) to your dependencies.",
+                        proxy,
+                    )
+
+                if use_socks_proxy:
+                    return SOCKSProxyManager(proxy, **opts)
+                else:
+                    return urllib3.PoolManager(**opts)
+            else:
+                return urllib3.ProxyManager(proxy, **opts)
+        else:
+            return urllib3.PoolManager(**opts)
+
+    def _request(
+        self: Self,
+        method: str,
+        endpoint_type: EndpointType,
+        body: Any,
+        headers: Mapping[str, str],
+    ) -> urllib3.BaseHTTPResponse:
+        return self._pool.request(
+            method,
+            self._auth.get_api_url(endpoint_type),
+            body=body,
+            headers=headers,
+        )
+
+
 if not ASYNC_TRANSPORT_ENABLED:
     # Sorry, no AsyncHttpTransport for you
-    AsyncHttpTransport = BaseHttpTransport
+    AsyncHttpTransport = HttpTransport
 
 else:
 
@@ -807,112 +913,6 @@ def kill(self: Self) -> Optional[asyncio.Task[None]]:  # type: ignore
                 return None
 
 
-class HttpTransport(BaseHttpTransport):
-    if TYPE_CHECKING:
-        _pool: Union[PoolManager, ProxyManager]
-
-    def _get_pool_options(self: Self) -> Dict[str, Any]:
-
-        num_pools = self.options.get("_experiments", {}).get("transport_num_pools")
-        options = {
-            "num_pools": 2 if num_pools is None else int(num_pools),
-            "cert_reqs": "CERT_REQUIRED",
-            "timeout": urllib3.Timeout(total=self.TIMEOUT),
-        }
-
-        socket_options: Optional[List[Tuple[int, int, int | bytes]]] = None
-
-        if self.options["socket_options"] is not None:
-            socket_options = self.options["socket_options"]
-
-        if self.options["keep_alive"]:
-            if socket_options is None:
-                socket_options = []
-
-            used_options = {(o[0], o[1]) for o in socket_options}
-            for default_option in KEEP_ALIVE_SOCKET_OPTIONS:
-                if (default_option[0], default_option[1]) not in used_options:
-                    socket_options.append(default_option)
-
-        if socket_options is not None:
-            options["socket_options"] = socket_options
-
-        options["ca_certs"] = (
-            self.options["ca_certs"]  # User-provided bundle from the SDK init
-            or os.environ.get("SSL_CERT_FILE")
-            or os.environ.get("REQUESTS_CA_BUNDLE")
-            or certifi.where()
-        )
-
-        options["cert_file"] = self.options["cert_file"] or os.environ.get(
-            "CLIENT_CERT_FILE"
-        )
-        options["key_file"] = self.options["key_file"] or os.environ.get(
-            "CLIENT_KEY_FILE"
-        )
-
-        return options
-
-    def _make_pool(self: Self) -> Union[PoolManager, ProxyManager]:
-        if self.parsed_dsn is None:
-            raise ValueError("Cannot create HTTP-based transport without valid DSN")
-
-        proxy = None
-        no_proxy = self._in_no_proxy(self.parsed_dsn)
-
-        # try HTTPS first
-        https_proxy = self.options["https_proxy"]
-        if self.parsed_dsn.scheme == "https" and (https_proxy != ""):
-            proxy = https_proxy or (not no_proxy and getproxies().get("https"))
-
-        # maybe fallback to HTTP proxy
-        http_proxy = self.options["http_proxy"]
-        if not proxy and (http_proxy != ""):
-            proxy = http_proxy or (not no_proxy and getproxies().get("http"))
-
-        opts = self._get_pool_options()
-
-        if proxy:
-            proxy_headers = self.options["proxy_headers"]
-            if proxy_headers:
-                opts["proxy_headers"] = proxy_headers
-
-            if proxy.startswith("socks"):
-                use_socks_proxy = True
-                try:
-                    # Check if PySocks dependency is available
-                    from urllib3.contrib.socks import SOCKSProxyManager
-                except ImportError:
-                    use_socks_proxy = False
-                    logger.warning(
-                        "You have configured a SOCKS proxy (%s) but support for SOCKS proxies is not installed. Disabling proxy support. Please add `PySocks` (or `urllib3` with the `[socks]` extra) to your dependencies.",
-                        proxy,
-                    )
-
-                if use_socks_proxy:
-                    return SOCKSProxyManager(proxy, **opts)
-                else:
-                    return urllib3.PoolManager(**opts)
-            else:
-                return urllib3.ProxyManager(proxy, **opts)
-        else:
-            return urllib3.PoolManager(**opts)
-
-    def _request(
-        self: Self,
-        method: str,
-        endpoint_type: EndpointType,
-        body: Any,
-        headers: Mapping[str, str],
-    ) -> urllib3.BaseHTTPResponse:
-        return self._pool.request(
-            method,
-            self._auth.get_api_url(endpoint_type),
-            body=body,
-            headers=headers,
-        )
-
-
 if not HTTP2_ENABLED:
     # Sorry, no Http2Transport for you
     class Http2Transport(HttpTransport):
@@ -1053,14 +1053,24 @@ def make_transport(options: Dict[str, Any]) -> Optional[Transport]:
 
     use_http2_transport = options.get("_experiments", {}).get("transport_http2", False)
     use_async_transport = options.get("_experiments", {}).get("transport_async", False)
+    async_integration = any(
+        integration.__class__.__name__ == "AsyncioIntegration"
+        for integration in options.get("integrations") or []
+    )
+
     # By default, we use the http transport class
     transport_cls: Type[Transport] = (
         Http2Transport if use_http2_transport else HttpTransport
     )
     if use_async_transport and ASYNC_TRANSPORT_ENABLED:
         try:
             asyncio.get_running_loop()
-            transport_cls = AsyncHttpTransport
+            if async_integration:
+                transport_cls = AsyncHttpTransport
+            else:
+                logger.warning(
+                    "You tried to use AsyncHttpTransport but the AsyncioIntegration is not enabled. Falling back to sync transport."
+                )
         except RuntimeError:
             # No event loop running, fall back to sync transport
             logger.warning("No event loop running, falling back to sync transport.")

tests/test_client.py

@@ -23,6 +23,8 @@
 from sentry_sdk.spotlight import DEFAULT_SPOTLIGHT_URL
 from sentry_sdk.utils import capture_internal_exception
 from sentry_sdk.integrations.executing import ExecutingIntegration
+from sentry_sdk.integrations.asyncio import AsyncioIntegration
+
 from sentry_sdk.transport import Transport, AsyncHttpTransport
 from sentry_sdk.serializer import MAX_DATABAG_BREADTH
 from sentry_sdk.consts import DEFAULT_MAX_BREADCRUMBS, DEFAULT_MAX_VALUE_LENGTH
@@ -1693,7 +1695,10 @@ async def test_async_proxy(monkeypatch, testcase):
     if testcase.get("env_no_proxy") is not None:
         monkeypatch.setenv("NO_PROXY", testcase["env_no_proxy"])
 
-    kwargs = {"_experiments": {"transport_async": True}}
+    kwargs = {
+        "_experiments": {"transport_async": True},
+        "integrations": [AsyncioIntegration()],
+    }
 
     if testcase["arg_http_proxy"] is not None:
         kwargs["http_proxy"] = testcase["arg_http_proxy"]
@@ -1795,7 +1800,10 @@ async def test_async_socks_proxy(testcase):
     # These are just the same tests as the sync ones, but they need to be run in an event loop
     # and respect the shutdown behavior of the async transport
 
-    kwargs = {"_experiments": {"transport_async": True}}
+    kwargs = {
+        "_experiments": {"transport_async": True},
+        "integrations": [AsyncioIntegration()],
+    }
 
     if testcase["arg_http_proxy"] is not None:
         kwargs["http_proxy"] = testcase["arg_http_proxy"]

tests/test_transport.py

@@ -33,6 +33,7 @@
     AsyncHttpTransport,
 )
 from sentry_sdk.integrations.logging import LoggingIntegration, ignore_logger
+from sentry_sdk.integrations.asyncio import AsyncioIntegration
 
 
 server = None
@@ -183,6 +184,7 @@ async def test_transport_works_async(
     client = make_client(
         debug=debug,
         _experiments=experiments,
+        integrations=[AsyncioIntegration()],
     )
 
     if use_pickle:
@@ -812,7 +814,7 @@ async def test_async_transport_background_thread_capture(
     """Test capture_envelope from background threads uses run_coroutine_threadsafe"""
     caplog.set_level(logging.DEBUG)
     experiments = {"transport_async": True}
-    client = make_client(_experiments=experiments)
+    client = make_client(_experiments=experiments, integrations=[AsyncioIntegration()])
     assert isinstance(client.transport, AsyncHttpTransport)
     sentry_sdk.get_global_scope().set_client(client)
     captured_from_thread = []
@@ -843,7 +845,7 @@ async def test_async_transport_event_loop_closed_scenario(
     """Test behavior when trying to capture after event loop context ends"""
     caplog.set_level(logging.DEBUG)
     experiments = {"transport_async": True}
-    client = make_client(_experiments=experiments)
+    client = make_client(_experiments=experiments, integrations=[AsyncioIntegration()])
     sentry_sdk.get_global_scope().set_client(client)
     original_loop = client.transport.loop
 
@@ -869,7 +871,7 @@ async def test_async_transport_concurrent_requests(
     """Test multiple simultaneous envelope submissions"""
     caplog.set_level(logging.DEBUG)
     experiments = {"transport_async": True}
-    client = make_client(_experiments=experiments)
+    client = make_client(_experiments=experiments, integrations=[AsyncioIntegration()])
     assert isinstance(client.transport, AsyncHttpTransport)
     sentry_sdk.get_global_scope().set_client(client)
 
@@ -891,7 +893,7 @@ async def test_async_transport_rate_limiting_with_concurrency(
 ):
     """Test async transport rate limiting with concurrent requests"""
     experiments = {"transport_async": True}
-    client = make_client(_experiments=experiments)
+    client = make_client(_experiments=experiments, integrations=[AsyncioIntegration()])
 
     assert isinstance(client.transport, AsyncHttpTransport)
     sentry_sdk.get_global_scope().set_client(client)
@@ -929,6 +931,7 @@ async def test_async_two_way_ssl_authentication():
         cert_file=cert_file,
         key_file=key_file,
         _experiments={"transport_async": True},
+        integrations=[AsyncioIntegration()],
     )
     assert isinstance(client.transport, AsyncHttpTransport)