feat: Add experimental async transport (port of PR #4572)

scripts/populate_tox/config.py

@@ -122,7 +122,7 @@
                 "pytest-asyncio",
                 "python-multipart",
                 "requests",
-                "anyio<4",
+                "anyio>=3,<5",
                 "jinja2",
             ],
             # There's an incompatibility between FastAPI's TestClient, which is
@@ -133,6 +133,7 @@
             # FastAPI versions we use older httpx which still supports the
             # deprecated argument.
             "<0.110.1": ["httpx<0.28.0"],
+            "<0.80": ["anyio<4"],
             "py3.6": ["aiocontextvars"],
         },
     },
@@ -171,7 +172,8 @@
     "httpx": {
         "package": "httpx",
         "deps": {
-            "*": ["anyio<4.0.0"],
+            "*": ["anyio>=3,<5"],
+            "<0.24": ["anyio<4"],
             ">=0.16,<0.17": ["pytest-httpx==0.10.0"],
             ">=0.17,<0.19": ["pytest-httpx==0.12.0"],
             ">=0.19,<0.21": ["pytest-httpx==0.14.0"],

scripts/populate_tox/tox.jinja

@@ -78,14 +78,17 @@
 
     linters: -r requirements-linting.txt
     linters: werkzeug<2.3.0
+    linters: httpcore[asyncio]
 
     mypy: -r requirements-linting.txt
     mypy: werkzeug<2.3.0
+    mypy: httpcore[asyncio]
     ruff: -r requirements-linting.txt
 
     # === Common ===
     py3.8-common: hypothesis
     common: pytest-asyncio
+    common: httpcore[asyncio]
     # See https://github.com/pytest-dev/pytest/issues/9621
     # and https://github.com/pytest-dev/pytest-forked/issues/67
     # for justification of the upper bound on pytest

sentry_sdk/__init__.py

@@ -25,6 +25,7 @@
     "configure_scope",
     "continue_trace",
     "flush",
+    "flush_async",
     "get_baggage",
     "get_client",
     "get_global_scope",

sentry_sdk/api.py

@@ -60,6 +60,7 @@ def overload(x: "T") -> "T":
     "configure_scope",
     "continue_trace",
     "flush",
+    "flush_async",
     "get_baggage",
     "get_client",
     "get_global_scope",
@@ -351,6 +352,14 @@ def flush(
     return get_client().flush(timeout=timeout, callback=callback)
 
 
+@clientmethod
+async def flush_async(
+    timeout: "Optional[float]" = None,
+    callback: "Optional[Callable[[int, float], None]]" = None,
+) -> None:
+    return await get_client().flush_async(timeout=timeout, callback=callback)
+
+
 @scopemethod
 def start_span(
     **kwargs: "Any",

sentry_sdk/client.py

@@ -31,7 +31,11 @@
 from sentry_sdk.serializer import serialize
 from sentry_sdk.tracing import trace
 from sentry_sdk.tracing_utils import has_span_streaming_enabled
-from sentry_sdk.transport import BaseHttpTransport, make_transport
+from sentry_sdk.transport import (
+    HttpTransportCore,
+    make_transport,
+    AsyncHttpTransport,
+)
 from sentry_sdk.consts import (
     SPANDATA,
     DEFAULT_MAX_VALUE_LENGTH,
@@ -251,6 +255,12 @@
     def flush(self, *args: "Any", **kwargs: "Any") -> None:
         return None
 
+    async def close_async(self, *args: "Any", **kwargs: "Any") -> None:
+        return None
+
+    async def flush_async(self, *args: "Any", **kwargs: "Any") -> None:
+        return None
+
     def __enter__(self) -> "BaseClient":
         return self
 
@@ -472,7 +482,7 @@
             or self.metrics_batcher
             or self.span_batcher
             or has_profiling_enabled(self.options)
-            or isinstance(self.transport, BaseHttpTransport)
+            or isinstance(self.transport, HttpTransportCore)
         ):
             # If we have anything on that could spawn a background thread, we
             # need to check if it's safe to use them.
@@ -999,6 +1009,32 @@
 
         return self.integrations.get(integration_name)
 
+    def _has_async_transport(self) -> bool:
+        """Check if the current transport is async."""
+        return isinstance(self.transport, AsyncHttpTransport)
+
+    @property
+    def _batchers(self) -> "tuple[Any, ...]":
+        return tuple(
+            b
+            for b in (self.log_batcher, self.metrics_batcher, self.span_batcher)
+            if b is not None
+        )
+
+    def _close_components(self) -> None:
+        """Kill all client components in the correct order."""
+        self.session_flusher.kill()
+        for b in self._batchers:
+            b.kill()
+        if self.monitor:
+            self.monitor.kill()
+
+    def _flush_components(self) -> None:
+        """Flush all client components."""
+        self.session_flusher.flush()
+        for b in self._batchers:
+            b.flush()
+
     def close(
         self,
         timeout: "Optional[float]" = None,
@@ -1009,19 +1045,40 @@
         semantics as :py:meth:`Client.flush`.
         """
         if self.transport is not None:
-            self.flush(timeout=timeout, callback=callback)
-            self.session_flusher.kill()
-            if self.log_batcher is not None:
-                self.log_batcher.kill()
-            if self.metrics_batcher is not None:
-                self.metrics_batcher.kill()
-            if self.span_batcher is not None:
-                self.span_batcher.kill()
-            if self.monitor:
-                self.monitor.kill()
+            if self._has_async_transport():
+                warnings.warn(
+                    "close() used with AsyncHttpTransport. Use close_async() instead.",
+                    stacklevel=2,
+                )
+                self._flush_components()
+            else:
+                self.flush(timeout=timeout, callback=callback)
+            self._close_components()
             self.transport.kill()
             self.transport = None
 
+    async def close_async(
+        self,
+        timeout: "Optional[float]" = None,
+        callback: "Optional[Callable[[int, float], None]]" = None,
+    ) -> None:
+        """
+        Asynchronously close the client and shut down the transport. Arguments have the same
+        semantics as :py:meth:`Client.flush_async`.
+        """
+        if self.transport is not None:
+            if not self._has_async_transport():
+                logger.debug(
+                    "close_async() used with non-async transport, aborting. Please use close() instead."
+                )
+                return
+            await self.flush_async(timeout=timeout, callback=callback)
+            self._close_components()
+            kill_task = self.transport.kill()  # type: ignore
+            if kill_task is not None:
+                await kill_task
+            self.transport = None
+
     def flush(
         self,
         timeout: "Optional[float]" = None,
@@ -1035,23 +1092,55 @@
         :param callback: Is invoked with the number of pending events and the configured timeout.
         """
         if self.transport is not None:
+            if self._has_async_transport():
+                warnings.warn(
+                    "flush() used with AsyncHttpTransport. Use flush_async() instead.",
+                    stacklevel=2,
+                )
+                return
             if timeout is None:
                 timeout = self.options["shutdown_timeout"]
-            self.session_flusher.flush()
-            if self.log_batcher is not None:
-                self.log_batcher.flush()
-            if self.metrics_batcher is not None:
-                self.metrics_batcher.flush()
-            if self.span_batcher is not None:
-                self.span_batcher.flush()
+            self._flush_components()
+
             self.transport.flush(timeout=timeout, callback=callback)
 
+    async def flush_async(
+        self,
+        timeout: "Optional[float]" = None,
+        callback: "Optional[Callable[[int, float], None]]" = None,
+    ) -> None:
+        """
+        Asynchronously wait for the current events to be sent.
+
+        :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used.
+
+        :param callback: Is invoked with the number of pending events and the configured timeout.
+        """
+        if self.transport is not None:
+            if not self._has_async_transport():
+                logger.debug(
+                    "flush_async() used with non-async transport, aborting. Please use flush() instead."
+                )
+                return
+            if timeout is None:
+                timeout = self.options["shutdown_timeout"]
+            self._flush_components()
+            flush_task = self.transport.flush(timeout=timeout, callback=callback)  # type: ignore
+            if flush_task is not None:
+                await flush_task
+
     def __enter__(self) -> "_Client":
         return self
 
     def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
         self.close()
 
+    async def __aenter__(self) -> "_Client":
+        return self
+
+    async def __aexit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
+        await self.close_async()
+
 
 from typing import TYPE_CHECKING
 

sentry_sdk/consts.py

@@ -78,6 +78,7 @@ class CompressionAlgo(Enum):
             "transport_compression_algo": Optional[CompressionAlgo],
             "transport_num_pools": Optional[int],
             "transport_http2": Optional[bool],
+            "transport_async": Optional[bool],
             "enable_logs": Optional[bool],
             "before_send_log": Optional[Callable[[Log, Hint], Optional[Log]]],
             "enable_metrics": Optional[bool],