ref(transport): Added shared sync/async transport superclass and created a sync transport HTTP subclass

Moved shared sync/async logic into a new superclass (HttpTransportCore), and moved sync transport specific code into a new subclass(BaseSyncHttpTransport), from which the current transport implementations inherit

Fixes GH-4568

Author: srothh

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 BaseHttpTransport, make_transport
+from sentry_sdk.transport import HttpTransportCore, make_transport
 from sentry_sdk.consts import (
     SPANDATA,
     DEFAULT_MAX_VALUE_LENGTH,
@@ -403,7 +403,7 @@ def _capture_envelope(envelope: Envelope) -> None:
             self.monitor
             or self.log_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.

sentry_sdk/transport.py

@@ -168,7 +168,7 @@ def _parse_rate_limits(
             continue
 
 
-class BaseHttpTransport(Transport):
+class HttpTransportCore(Transport):
     """The base HTTP transport."""
 
     TIMEOUT = 30  # seconds
@@ -292,12 +292,8 @@ def _update_rate_limits(
                 seconds=retry_after
             )
 
-    def _send_request(
-        self: Self,
-        body: bytes,
-        headers: Dict[str, str],
-        endpoint_type: EndpointType = EndpointType.ENVELOPE,
-        envelope: Optional[Envelope] = None,
+    def _handle_request_error(
+        self: Self, envelope: Optional[Envelope], loss_reason: str = "network"
     ) -> None:
         def record_loss(reason: str) -> None:
             if envelope is None:
@@ -306,45 +302,45 @@ def record_loss(reason: str) -> None:
                 for item in envelope.items:
                     self.record_lost_event(reason, item=item)
 
+        self.on_dropped_event(loss_reason)
+        record_loss("network_error")
+
+    def _handle_response(
+        self: Self,
+        response: Union[urllib3.BaseHTTPResponse, httpcore.Response],
+        envelope: Optional[Envelope],
+    ) -> None:
+        self._update_rate_limits(response)
+
+        if response.status == 429:
+            # if we hit a 429.  Something was rate limited but we already
+            # acted on this in `self._update_rate_limits`.  Note that we
+            # do not want to record event loss here as we will have recorded
+            # an outcome in relay already.
+            self.on_dropped_event("status_429")
+            pass
+
+        elif response.status >= 300 or response.status < 200:
+            logger.error(
+                "Unexpected status code: %s (body: %s)",
+                response.status,
+                getattr(response, "data", getattr(response, "content", None)),
+            )
+            self._handle_request_error(
+                envelope=envelope, loss_reason="status_{}".format(response.status)
+            )
+
+    def _update_headers(
+        self: Self,
+        headers: Dict[str, str],
+    ) -> None:
+
         headers.update(
             {
                 "User-Agent": str(self._auth.client),
                 "X-Sentry-Auth": str(self._auth.to_header()),
             }
         )
-        try:
-            response = self._request(
-                "POST",
-                endpoint_type,
-                body,
-                headers,
-            )
-        except Exception:
-            self.on_dropped_event("network")
-            record_loss("network_error")
-            raise
-
-        try:
-            self._update_rate_limits(response)
-
-            if response.status == 429:
-                # if we hit a 429.  Something was rate limited but we already
-                # acted on this in `self._update_rate_limits`.  Note that we
-                # do not want to record event loss here as we will have recorded
-                # an outcome in relay already.
-                self.on_dropped_event("status_429")
-                pass
-
-            elif response.status >= 300 or response.status < 200:
-                logger.error(
-                    "Unexpected status code: %s (body: %s)",
-                    response.status,
-                    getattr(response, "data", getattr(response, "content", None)),
-                )
-                self.on_dropped_event("status_{}".format(response.status))
-                record_loss("network_error")
-        finally:
-            response.close()
 
     def on_dropped_event(self: Self, _reason: str) -> None:
         return None
@@ -381,11 +377,6 @@ def _fetch_pending_client_report(
             type="client_report",
         )
 
-    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]))
-
     def _check_disabled(self: Self, category: EventDataCategory) -> bool:
         def _disabled(bucket: Optional[EventDataCategory]) -> bool:
             ts = self._disabled_until.get(bucket)
@@ -404,9 +395,9 @@ def _is_worker_full(self: Self) -> bool:
     def is_healthy(self: Self) -> bool:
         return not (self._is_worker_full() or self._is_rate_limited())
 
-    def _send_envelope(self: Self, envelope: Envelope) -> None:
-
-        # remove all items from the envelope which are over quota
+    def _prepare_envelope(
+        self: Self, envelope: Envelope
+    ) -> Optional[Tuple[Envelope, io.BytesIO, Dict[str, str]]]:
         new_items = []
         for item in envelope.items:
             if self._check_disabled(item.data_category):
@@ -448,13 +439,7 @@ def _send_envelope(self: Self, envelope: Envelope) -> None:
         if content_encoding:
             headers["Content-Encoding"] = content_encoding
 
-        self._send_request(
-            body.getvalue(),
-            headers=headers,
-            endpoint_type=EndpointType.ENVELOPE,
-            envelope=envelope,
-        )
-        return None
+        return envelope, body, headers
 
     def _serialize_envelope(
         self: Self, envelope: Envelope
@@ -512,6 +497,54 @@ def _request(
     ) -> Union[urllib3.BaseHTTPResponse, httpcore.Response]:
         raise NotImplementedError()
 
+    def kill(self: Self) -> None:
+        logger.debug("Killing HTTP transport")
+        self._worker.kill()
+
+
+class BaseSyncHttpTransport(HttpTransportCore):
+
+    def _send_envelope(self: Self, envelope: Envelope) -> None:
+        _prepared_envelope = self._prepare_envelope(envelope)
+        if _prepared_envelope is None:  # TODO: check this behaviour in detail
+            return None
+        envelope, body, headers = _prepared_envelope
+        self._send_request(
+            body.getvalue(),
+            headers=headers,
+            endpoint_type=EndpointType.ENVELOPE,
+            envelope=envelope,
+        )
+        return None
+
+    def _send_request(
+        self: Self,
+        body: bytes,
+        headers: Dict[str, str],
+        endpoint_type: EndpointType,
+        envelope: Optional[Envelope],
+    ) -> None:
+        self._update_headers(headers)
+        try:
+            response = 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()
+
+    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]))
+
     def capture_envelope(self: Self, envelope: Envelope) -> None:
         def send_envelope_wrapper() -> None:
             with capture_internal_exceptions():
@@ -534,12 +567,8 @@ def flush(
             self._worker.submit(lambda: self._flush_client_reports(force=True))
             self._worker.flush(timeout, callback)
 
-    def kill(self: Self) -> None:
-        logger.debug("Killing HTTP transport")
-        self._worker.kill()
-
 
-class HttpTransport(BaseHttpTransport):
+class HttpTransport(BaseSyncHttpTransport):
     if TYPE_CHECKING:
         _pool: Union[PoolManager, ProxyManager]
 
@@ -656,7 +685,7 @@ def __init__(self: Self, options: Dict[str, Any]) -> None:
 
 else:
 
-    class Http2Transport(BaseHttpTransport):  # type: ignore
+    class Http2Transport(BaseSyncHttpTransport):  # type: ignore
         """The HTTP2 transport based on httpcore."""
 
         TIMEOUT = 15