feat(transport): Add async transport class

Add an implementation of Transport to work with the async background worker and HTTPCore async.

GH-4582

Author: srothh

sentry_sdk/transport.py

@@ -6,6 +6,7 @@
 import socket
 import ssl
 import time
+import asyncio
 from datetime import datetime, timedelta, timezone
 from collections import defaultdict
 from urllib.request import getproxies
@@ -571,6 +572,187 @@ def flush(
             self._worker.flush(timeout, callback)
 
 
+class AsyncHttpTransport(HttpTransportCore):
+    def __init__(self: Self, options: Dict[str, Any]) -> None:
+        super().__init__(options)
+        # Requires event loop at init time
+        self._loop = asyncio.get_running_loop()
+        self.background_tasks = set()
+
+    async def _send_envelope(self: Self, envelope: Envelope) -> None:
+        _prepared_envelope = self._prepare_envelope(envelope)
+        if _prepared_envelope is None:
+            return None
+        envelope, body, headers = _prepared_envelope
+        await self._send_request(
+            body.getvalue(),
+            headers=headers,
+            endpoint_type=EndpointType.ENVELOPE,
+            envelope=envelope,
+        )
+        return None
+
+    async def _send_request(
+        self: Self,
+        body: bytes,
+        headers: Dict[str, str],
+        endpoint_type: EndpointType,
+        envelope: Optional[Envelope],
+    ) -> None:
+        self._update_headers(headers)
+        try:
+            response = await self._request(
+                "POST",
+                endpoint_type,
+                body,
+                headers,
+            )
+        except Exception:
+            self._handle_request_error(envelope=envelope, loss_reason="network")
+            raise
+        try:
+            self._handle_response(response=response, envelope=envelope)
+        finally:
+            response.close()
+
+    async def _request(
+        self: Self,
+        method: str,
+        endpoint_type: EndpointType,
+        body: Any,
+        headers: Mapping[str, str],
+    ) -> httpcore.Response:
+        return await self._pool.request(
+            method,
+            self._auth.get_api_url(endpoint_type),
+            content=body,
+            headers=headers,  # type: ignore
+        )
+
+    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:
+            self.capture_envelope(Envelope(items=[client_report]))
+
+    async def _capture_envelope(self: Self, envelope: Envelope) -> None:
+        async def send_envelope_wrapper() -> None:
+            with capture_internal_exceptions():
+                await self._send_envelope(envelope)
+                self._flush_client_reports()
+
+        if not self._worker.submit(send_envelope_wrapper):
+            self.on_dropped_event("full_queue")
+            for item in envelope.items:
+                self.record_lost_event("queue_overflow", item=item)
+
+    def capture_envelope(self: Self, envelope: Envelope) -> None:
+        # Synchronous entry point
+        if asyncio.get_running_loop() is not None:
+            # 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)
+        else:
+            # We are in a background thread, not running an event loop,
+            # have to launch the task on the loop in a threadsafe way.
+            asyncio.run_coroutine_threadsafe(
+                self._capture_envelope(envelope),
+                self._loop,
+            )
+
+    async def flush_async(
+        self: Self,
+        timeout: float,
+        callback: Optional[Callable[[int, float], None]] = None,
+    ) -> None:
+        logger.debug("Flushing HTTP transport")
+
+        if timeout > 0:
+            self._worker.submit(lambda: self._flush_client_reports(force=True))
+            await self._worker.flush_async(timeout, callback)  # type: ignore
+
+    def _get_pool_options(self: Self) -> Dict[str, Any]:
+        options: Dict[str, Any] = {
+            "http2": False,  # no HTTP2 for now
+            "retries": 3,
+        }
+
+        socket_options = (
+            self.options["socket_options"]
+            if self.options["socket_options"] is not None
+            else []
+        )
+
+        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)
+
+        options["socket_options"] = socket_options
+
+        ssl_context = ssl.create_default_context()
+        ssl_context.load_verify_locations(
+            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()
+        )
+        cert_file = self.options["cert_file"] or os.environ.get("CLIENT_CERT_FILE")
+        key_file = self.options["key_file"] or os.environ.get("CLIENT_KEY_FILE")
+        if cert_file is not None:
+            ssl_context.load_cert_chain(cert_file, key_file)
+
+        options["ssl_context"] = ssl_context
+
+        return options
+
+    def _make_pool(
+        self: Self,
+    ) -> Union[
+        httpcore.AsyncSOCKSProxy, httpcore.AsyncHTTPProxy, httpcore.AsyncConnectionPool
+    ]:
+        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"):
+                try:
+                    if "socket_options" in opts:
+                        socket_options = opts.pop("socket_options")
+                        if socket_options:
+                            logger.warning(
+                                "You have defined socket_options but using a SOCKS proxy which doesn't support these. We'll ignore socket_options."
+                            )
+                    return httpcore.AsyncSOCKSProxy(proxy_url=proxy, **opts)
+                except RuntimeError:
+                    logger.warning(
+                        "You have configured a SOCKS proxy (%s) but support for SOCKS proxies is not installed. Disabling proxy support.",
+                        proxy,
+                    )
+            else:
+                return httpcore.AsyncHTTPProxy(proxy_url=proxy, **opts)
+
+        return httpcore.AsyncConnectionPool(**opts)
+
+
 class HttpTransport(BaseHttpTransport):
     if TYPE_CHECKING:
         _pool: Union[PoolManager, ProxyManager]