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

Author: srothh

sentry_sdk/transport.py

@@ -238,17 +238,7 @@ def __init__(self: Self, options: Dict[str, Any]) -> None:
             self._compression_level = 4
 
     def _create_worker(self, options: dict[str, Any]) -> Worker:
-        async_enabled = options.get("_experiments", {}).get("transport_async", False)
-        try:
-            asyncio.get_running_loop()
-            worker_cls = (
-                AsyncWorker
-                if async_enabled and ASYNC_TRANSPORT_ENABLED
-                else BackgroundWorker
-            )
-        except RuntimeError:
-            worker_cls = BackgroundWorker
-        return worker_cls(queue_size=options["transport_queue_size"])
+        raise NotImplementedError()
 
     def record_lost_event(
         self: Self,
@@ -565,6 +555,9 @@ def _send_request(
         finally:
             response.close()
 
+    def _create_worker(self: Self, options: dict[str, Any]) -> Worker:
+        return BackgroundWorker(queue_size=options["transport_queue_size"])
+
     def _flush_client_reports(self: Self, force: bool = False) -> None:
         client_report = self._fetch_pending_client_report(force=force, interval=60)
         if client_report is not None:
@@ -595,12 +588,7 @@ def flush(
 
 if not ASYNC_TRANSPORT_ENABLED:
     # Sorry, no AsyncHttpTransport for you
-    class AsyncHttpTransport(BaseHttpTransport):
-        def __init__(self: Self, options: Dict[str, Any]) -> None:
-            super().__init__(options)
-            logger.warning(
-                "You tried to use AsyncHttpTransport but don't have httpcore[asyncio] installed. Falling back to sync transport."
-            )
+    AsyncHttpTransport = BaseHttpTransport
 
 else:
 
@@ -611,6 +599,9 @@ def __init__(self: Self, options: Dict[str, Any]) -> None:
             self.loop = asyncio.get_running_loop()
             self.background_tasks: set[asyncio.Task[None]] = set()
 
+        def _create_worker(self: Self, options: dict[str, Any]) -> Worker:
+            return AsyncWorker(queue_size=options["transport_queue_size"])
+
         def _get_header_value(self: Self, response: Any, header: str) -> Optional[str]:
             return next(
                 (
@@ -683,7 +674,7 @@ async def _flush_client_reports(self: Self, force: bool = False) -> None:
             if client_report is not None:
                 self.capture_envelope(Envelope(items=[client_report]))
 
-        async def _capture_envelope(self: Self, envelope: Envelope) -> None:
+        def _capture_envelope(self: Self, envelope: Envelope) -> None:
             async def send_envelope_wrapper() -> None:
                 with capture_internal_exceptions():
                     await self._send_envelope(envelope)
@@ -696,26 +687,14 @@ async def send_envelope_wrapper() -> None:
 
         def capture_envelope(self: Self, envelope: Envelope) -> None:
             # Synchronous entry point
-            try:
-                asyncio.get_running_loop()
-                # We are on the main thread running the event loop
-                task = asyncio.create_task(self._capture_envelope(envelope))
-                self.background_tasks.add(task)
-                task.add_done_callback(self.background_tasks.discard)
-            except RuntimeError:
-                # We are in a background thread, not running an event loop,
-                # have to launch the task on the loop in a threadsafe way.
-                if self.loop and self.loop.is_running():
-                    asyncio.run_coroutine_threadsafe(
-                        self._capture_envelope(envelope),
-                        self.loop,
-                    )
-                else:
-                    # The event loop is no longer running
-                    logger.warning("Async Transport is not running in an event loop.")
-                    self.on_dropped_event("internal_sdk_error")
-                    for item in envelope.items:
-                        self.record_lost_event("internal_sdk_error", item=item)
+            if self.loop and self.loop.is_running():
+                self.loop.call_soon_threadsafe(self._capture_envelope, envelope)
+            else:
+                # The event loop is no longer running
+                logger.warning("Async Transport is not running in an event loop.")
+                self.on_dropped_event("internal_sdk_error")
+                for item in envelope.items:
+                    self.record_lost_event("internal_sdk_error", item=item)
 
         def flush(  # type: ignore[override]
             self: Self,
@@ -1074,16 +1053,20 @@ 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)
     # By default, we use the http transport class
-    if use_async_transport:
+    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: Type[Transport] = AsyncHttpTransport
+            transport_cls = AsyncHttpTransport
         except RuntimeError:
             # No event loop running, fall back to sync transport
             logger.warning("No event loop running, falling back to sync transport.")
-            transport_cls = Http2Transport if use_http2_transport else HttpTransport
-    else:
-        transport_cls = Http2Transport if use_http2_transport else HttpTransport
+    elif use_async_transport:
+        logger.warning(
+            "You tried to use AsyncHttpTransport but don't have httpcore[asyncio] installed. Falling back to sync transport."
+        )
 
     if isinstance(ref_transport, Transport):
         return ref_transport

setup.py

@@ -60,6 +60,7 @@ def get_file_text(file_name):
         "flask": ["flask>=0.11", "blinker>=1.1", "markupsafe"],
         "grpcio": ["grpcio>=1.21.1", "protobuf>=3.8.0"],
         "http2": ["httpcore[http2]==1.*"],
+        "asyncio": ["httpcore[asyncio]==1.*"],
         "httpx": ["httpx>=0.16.0"],
         "huey": ["huey>=2"],
         "huggingface_hub": ["huggingface_hub>=0.22"],