Integrate async transport with SDK

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/client.py

@@ -3,6 +3,7 @@
 import uuid
 import random
 import socket
+import asyncio
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from importlib import import_module
@@ -25,7 +26,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,
@@ -917,51 +918,126 @@ def get_integration(
 
         return self.integrations.get(integration_name)
 
-    def close(
+    def _close_transport(self) -> Optional[asyncio.Task[None]]:
+        """Close transport and return cleanup task if any."""
+        if self.transport is not None:
+            cleanup_task = self.transport.kill()  # type: ignore
+            self.transport = None
+            return cleanup_task
+        return None
+
+    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()
+
+    async def _close_components_async(self) -> None:
+        """Async version of _close_components that properly awaits transport cleanup."""
+        self._close_components()
+        cleanup_task = self._close_transport()
+        if cleanup_task is not None:
+            await cleanup_task
+
+    def close(  # type: ignore[override]
         self,
         timeout: Optional[float] = None,
         callback: Optional[Callable[[int, float], None]] = None,
-    ) -> None:
+    ) -> Optional[asyncio.Task[None]]:
         """
         Close the client and shut down the transport. Arguments have the same
-        semantics as :py:meth:`Client.flush`.
+        semantics as :py:meth:`Client.flush`. When using the async transport, close needs to be awaited to block.
         """
-        if self.transport is not None:
-            self.flush(timeout=timeout, callback=callback)
 
-            self.session_flusher.kill()
+        async def _flush_and_close(
+            timeout: Optional[float], callback: Optional[Callable[[int, float], None]]
+        ) -> None:
 
-            if self.log_batcher is not None:
-                self.log_batcher.kill()
+            await self._flush_async(timeout=timeout, callback=callback)
+            await self._close_components_async()
 
-            if self.monitor:
-                self.monitor.kill()
+        if self.transport is not None:
+            if isinstance(self.transport, AsyncHttpTransport) and hasattr(
+                self.transport, "loop"
+            ):
 
-            self.transport.kill()
-            self.transport = None
+                try:
+                    flush_task = self.transport.loop.create_task(
+                        _flush_and_close(timeout, callback)
+                    )
+                except RuntimeError:
+                    # Shutdown the components anyway
+                    self._close_components()
+                    self._close_transport()
+                    logger.warning("Event loop not running, aborting close.")
+                    return None
+                # Enforce flush before shutdown
+                return flush_task
+            else:
+                self.flush(timeout=timeout, callback=callback)
+                self._close_components()
+                self._close_transport()
+
+        return None
 
-    def flush(
+    def flush(  # type: ignore[override]
         self,
         timeout: Optional[float] = None,
         callback: Optional[Callable[[int, float], None]] = None,
-    ) -> None:
+    ) -> Optional[asyncio.Task[None]]:
         """
-        Wait for the current events to be sent.
+        Wait for the current events to be sent. When using the async transport, flush needs to be awaited to block.
 
         :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 timeout is None:
-                timeout = self.options["shutdown_timeout"]
-            self.session_flusher.flush()
+            if isinstance(self.transport, AsyncHttpTransport) and hasattr(
+                self.transport, "loop"
+            ):
+                try:
+                    return self.transport.loop.create_task(
+                        self._flush_async(timeout, callback)
+                    )
+                except RuntimeError:
+                    logger.warning("Event loop not running, aborting flush.")
+                    return None
+            else:
+                self._flush_sync(timeout, callback)
+        return None
 
-            if self.log_batcher is not None:
-                self.log_batcher.flush()
+    def _flush_sync(
+        self, timeout: Optional[float], callback: Optional[Callable[[int, float], None]]
+    ) -> None:
+        """Synchronous flush implementation."""
+        if timeout is None:
+            timeout = self.options["shutdown_timeout"]
 
+        self._flush_components()
+        if self.transport is not None:
             self.transport.flush(timeout=timeout, callback=callback)
 
+    async def _flush_async(
+        self, timeout: Optional[float], callback: Optional[Callable[[int, float], None]]
+    ) -> None:
+        """Asynchronous flush implementation."""
+        if timeout is None:
+            timeout = self.options["shutdown_timeout"]
+
+        self._flush_components()
+        if self.transport is not None:
+            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,47 @@ 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
+
+            task = client.close()  # type: ignore
+            if task is not None:
+                await task
+        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 +166,4 @@ class AsyncioIntegration(Integration):
     @staticmethod
     def setup_once() -> None:
         patch_asyncio()
+        patch_loop_close()

sentry_sdk/transport.py

@@ -7,6 +7,7 @@
 import ssl
 import time
 import asyncio
+
 from datetime import datetime, timedelta, timezone
 from collections import defaultdict
 from urllib.request import getproxies
@@ -29,6 +30,8 @@
     HTTP2_ENABLED = False
 
 try:
+    import anyio  # noqa: F401
+
     ASYNC_TRANSPORT_ENABLED = httpcore is not None
 except ImportError:
     ASYNC_TRANSPORT_ENABLED = False

tests/integrations/asyncio/test_asyncio.py

@@ -377,3 +377,52 @@ async def test_span_origin(
 
     assert event["contexts"]["trace"]["origin"] == "manual"
     assert event["spans"][0]["origin"] == "auto.function.asyncio"
+
+
+@minimum_python_38
+def test_loop_close_patching(sentry_init):
+    sentry_init(integrations=[AsyncioIntegration()])
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    try:
+        with patch("asyncio.get_running_loop", return_value=loop):
+            assert not hasattr(loop, "_sentry_flush_patched")
+            AsyncioIntegration.setup_once()
+            assert hasattr(loop, "_sentry_flush_patched")
+            assert loop._sentry_flush_patched is True
+
+    finally:
+        if not loop.is_closed():
+            loop.close()
+
+
+@minimum_python_38
+def test_loop_close_flushes_async_transport(sentry_init):
+    from sentry_sdk.transport import AsyncHttpTransport
+    from unittest.mock import Mock, AsyncMock
+
+    sentry_init(integrations=[AsyncioIntegration()])
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    try:
+        with patch("asyncio.get_running_loop", return_value=loop):
+            AsyncioIntegration.setup_once()
+
+        mock_client = Mock()
+        mock_transport = Mock(spec=AsyncHttpTransport)
+        mock_client.transport = mock_transport
+        mock_client.close = AsyncMock(return_value=None)
+
+        with patch("sentry_sdk.get_client", return_value=mock_client):
+            loop.close()
+
+        mock_client.close.assert_called_once()
+        mock_client.close.assert_awaited_once()
+
+    except Exception:
+        if not loop.is_closed():
+            loop.close()