Integrate async transport with SDK (#4615)

Integrate the async transport with the rest of the SDK. Provide a new
experimental option `transport_async` that enables the async transport
if an event loop is running. Otherwise, fall back to the sync transport.

Furthermore, adapt the client to work with the async transport. To this
end, flush and close were changed to be non blocking and awaitable in an
async context to avoid deadlocks, however close enforces a completed
flush before shutdown. As there are to my knowledge no background
threads running flush/close, these methods are currently not
thread-safe/loop-aware for async, which can be changed if necessary.

Atexit issue: The atexit integration used by the SDK runs after the
event loop has already closed if asyncio.run() is used. This makes it
impossible for the async flush to happen, as atexit calls
client.close(), but a loop is no longer present. I attempted to apply
[this fix](https://discuss.python.org/t/atexit-for-asyncio/13911) by
patching the loop close in the asyncio integration, but I am unsure if I
did it correctly/put it in the correct spot, or if this is a good idea.
From my SDK test however, it seems to fix the flush issue. Note also
that this will apparently be patched in Python 3.14, as per the
discussion in the linked thread.

As a final note, I added event loop checking. Whenever the event loop is
used, the transport/client catch RuntimeErrors, which would be thrown in
case the event loop was already shut down. I am not sure if this is a
case we need to consider, but I added it for now because I did not want
the transport to potentially throw RuntimeError if the event loop is
shutdown during a program. If this should be left out currently for
simplicity, I can remove it again.

I added the [httpcore[asyncio] ](https://www.encode.io/httpcore/async/)
dependency to requirements-testing, as it is needed for the async
httpcore functionality.

GH-4601

---------

Co-authored-by: Neel Shah <[email protected]>

Author: srothh

requirements-testing.txt

@@ -11,7 +11,7 @@ asttokens
 responses
 pysocks
 socksio
-httpcore[http2]
+httpcore[http2,asyncio]
 setuptools
 freezegun
 Brotli

scripts/populate_tox/config.py

@@ -96,7 +96,7 @@
                 "pytest-asyncio",
                 "python-multipart",
                 "requests",
-                "anyio<4",
+                "anyio>=3,<5",
             ],
             # There's an incompatibility between FastAPI's TestClient, which is
             # actually Starlette's TestClient, which is actually httpx's Client.
@@ -106,6 +106,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"],
         },
     },

scripts/populate_tox/tox.jinja

@@ -207,7 +207,7 @@ deps =
     httpx-v0.25: pytest-httpx==0.25.0
     httpx: pytest-httpx
     # anyio is a dep of httpx
-    httpx: anyio<4.0.0
+    httpx: anyio>=3,<5
     httpx-v0.16: httpx~=0.16.0
     httpx-v0.18: httpx~=0.18.0
     httpx-v0.20: httpx~=0.20.0

sentry_sdk/api.py

@@ -226,6 +226,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)
+
+
 def start_span(**kwargs: Any) -> Span:
     """
     Start and return a span.

sentry_sdk/client.py

@@ -25,7 +25,7 @@
 )
 from sentry_sdk.serializer import serialize
 from sentry_sdk.tracing import trace
-from sentry_sdk.transport import HttpTransportCore, make_transport
+from sentry_sdk.transport import HttpTransportCore, make_transport, AsyncHttpTransport
 from sentry_sdk.consts import (
     SPANDATA,
     DEFAULT_MAX_VALUE_LENGTH,
@@ -214,6 +214,12 @@ def close(self, *args: Any, **kwargs: Any) -> None:
     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
 
@@ -917,6 +923,14 @@ def get_integration(
 
         return self.integrations.get(integration_name)
 
+    def _close_components(self) -> None:
+        """Kill all client components in the correct order."""
+        self.session_flusher.kill()
+        if self.log_batcher is not None:
+            self.log_batcher.kill()
+        if self.monitor:
+            self.monitor.kill()
+
     def close(
         self,
         timeout: Optional[float] = None,
@@ -927,19 +941,43 @@ def close(
         semantics as :py:meth:`Client.flush`.
         """
         if self.transport is not None:
+            if isinstance(self.transport, AsyncHttpTransport) and hasattr(
+                self.transport, "loop"
+            ):
+                logger.debug(
+                    "close() used with AsyncHttpTransport, aborting. Please use close_async() instead."
+                )
+                return
             self.flush(timeout=timeout, callback=callback)
-
-            self.session_flusher.kill()
-
-            if self.log_batcher is not None:
-                self.log_batcher.kill()
-
-            if self.monitor:
-                self.monitor.kill()
-
+            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 (
+                isinstance(self.transport, AsyncHttpTransport)
+                and hasattr(self.transport, "loop")
+            ):
+                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,
@@ -953,15 +991,52 @@ def flush(
         :param callback: Is invoked with the number of pending events and the configured timeout.
         """
         if self.transport is not None:
+            if isinstance(self.transport, AsyncHttpTransport) and hasattr(
+                self.transport, "loop"
+            ):
+                logger.debug(
+                    "flush() used with AsyncHttpTransport, aborting. Please use flush_async() instead."
+                )
+                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()
+            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 (
+                isinstance(self.transport, AsyncHttpTransport)
+                and hasattr(self.transport, "loop")
+            ):
+                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 _flush_components(self) -> None:
+        self.session_flusher.flush()
+        if self.log_batcher is not None:
+            self.log_batcher.flush()
+
     def __enter__(self) -> _Client:
         return self
 

sentry_sdk/consts.py

@@ -75,6 +75,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]]],
         },

sentry_sdk/integrations/asyncio.py

@@ -5,6 +5,7 @@
 from sentry_sdk.consts import OP
 from sentry_sdk.integrations import Integration, DidNotEnable
 from sentry_sdk.utils import event_from_exception, logger, reraise
+from sentry_sdk.transport import AsyncHttpTransport
 
 try:
     import asyncio
@@ -29,6 +30,45 @@ def get_name(coro: Any) -> str:
     )
 
 
+def patch_loop_close() -> None:
+    """Patch loop.close to flush pending events before shutdown."""
+    # Atexit shutdown hook happens after the event loop is closed.
+    # Therefore, it is necessary to patch the loop.close method to ensure
+    # that pending events are flushed before the interpreter shuts down.
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        # No running loop → cannot patch now
+        return
+
+    if getattr(loop, "_sentry_flush_patched", False):
+        return
+
+    async def _flush() -> None:
+        client = sentry_sdk.get_client()
+        if not client:
+            return
+
+        try:
+            if not isinstance(client.transport, AsyncHttpTransport):
+                return
+
+            await client.close_async()
+        except Exception:
+            logger.warning("Sentry flush failed during loop shutdown", exc_info=True)
+
+    orig_close = loop.close
+
+    def _patched_close() -> None:
+        try:
+            loop.run_until_complete(_flush())
+        finally:
+            orig_close()
+
+    loop.close = _patched_close  # type: ignore
+    loop._sentry_flush_patched = True  # type: ignore
+
+
 def patch_asyncio() -> None:
     orig_task_factory = None
     try:
@@ -124,3 +164,4 @@ class AsyncioIntegration(Integration):
     @staticmethod
     def setup_once() -> None:
         patch_asyncio()
+        patch_loop_close()