refactor: Simplify and deduplicate async transport code

Major cleanup that removes ~390 lines while preserving all functionality:

Architecture:
- Always define AsyncHttpTransport as a real class (not aliased to
  HttpTransport when deps missing). Raises RuntimeError in __init__
  instead. This eliminates the isinstance aliasing bug and removes
  most type: ignore annotations.
- Add _is_async_transport() helper to _Client, replacing 4 repeated
  isinstance checks.

Deduplication in transport.py:
- Extract _get_httpcore_pool_options() into HttpTransportCore — shared
  by AsyncHttpTransport and Http2Transport (SSL, certs, socket opts).
- Extract _resolve_proxy() into HttpTransportCore — shared proxy URL
  resolution logic.
- Extract _timeout_extensions property — shared timeout config dict.
- Extract _get_httpcore_header_value() module function — shared
  case-insensitive header lookup for httpcore responses.
- Enable HTTP/2 in AsyncHttpTransport when DSN scheme is HTTPS.

Simplification in client.py:
- Use _batchers tuple property in _flush_components/_close_components
  to avoid repeating the same None-check pattern for each batcher.

Test cleanup:
- Extract PROXY_TESTCASES and SOCKS_PROXY_TESTCASES constants shared
  between sync and async proxy tests (-300 lines).
- Add _make_async_transport_options() helper for make_transport tests.
- Remove section separator comments.
- Trim Worker ABC docstrings to one-liners.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: BYK

sentry_sdk/client.py

@@ -32,7 +32,6 @@
 from sentry_sdk.tracing import trace
 from sentry_sdk.tracing_utils import has_span_streaming_enabled
 from sentry_sdk.transport import (
-    ASYNC_TRANSPORT_ENABLED,
     HttpTransportCore,
     make_transport,
     AsyncHttpTransport,
@@ -1010,27 +1009,31 @@ def get_integration(
 
         return self.integrations.get(integration_name)
 
+    def _is_async_transport(self) -> bool:
+        """Check if the current transport is async."""
+        return isinstance(self.transport, AsyncHttpTransport)
+
+    @property
+    def _batchers(self) -> "tuple[Any, ...]":
+        return tuple(
+            b
+            for b in (self.log_batcher, self.metrics_batcher, self.span_batcher)
+            if b is not 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.metrics_batcher is not None:
-            self.metrics_batcher.kill()
-        if self.span_batcher is not None:
-            self.span_batcher.kill()
+        for b in self._batchers:
+            b.kill()
         if self.monitor:
             self.monitor.kill()
 
     def _flush_components(self) -> None:
         """Flush all client components."""
         self.session_flusher.flush()
-        if self.log_batcher is not None:
-            self.log_batcher.flush()
-        if self.metrics_batcher is not None:
-            self.metrics_batcher.flush()
-        if self.span_batcher is not None:
-            self.span_batcher.flush()
+        for b in self._batchers:
+            b.flush()
 
     def close(
         self,
@@ -1042,9 +1045,7 @@ def close(
         semantics as :py:meth:`Client.flush`.
         """
         if self.transport is not None:
-            if ASYNC_TRANSPORT_ENABLED and isinstance(
-                self.transport, AsyncHttpTransport
-            ):
+            if self._is_async_transport():
                 logger.warning(
                     "close() used with AsyncHttpTransport. "
                     "Prefer close_async() for graceful async shutdown. "
@@ -1067,10 +1068,7 @@ async def close_async(
         semantics as :py:meth:`Client.flush_async`.
         """
         if self.transport is not None:
-            if not (
-                ASYNC_TRANSPORT_ENABLED
-                and isinstance(self.transport, AsyncHttpTransport)
-            ):
+            if not self._is_async_transport():
                 logger.debug(
                     "close_async() used with non-async transport, aborting. Please use close() instead."
                 )
@@ -1095,9 +1093,7 @@ def flush(
         :param callback: Is invoked with the number of pending events and the configured timeout.
         """
         if self.transport is not None:
-            if ASYNC_TRANSPORT_ENABLED and isinstance(
-                self.transport, AsyncHttpTransport
-            ):
+            if self._is_async_transport():
                 logger.warning(
                     "flush() used with AsyncHttpTransport. Please use flush_async() instead."
                 )
@@ -1121,10 +1117,7 @@ async def flush_async(
         :param callback: Is invoked with the number of pending events and the configured timeout.
         """
         if self.transport is not None:
-            if not (
-                ASYNC_TRANSPORT_ENABLED
-                and isinstance(self.transport, AsyncHttpTransport)
-            ):
+            if not self._is_async_transport():
                 logger.debug(
                     "flush_async() used with non-async transport, aborting. Please use flush() instead."
                 )

sentry_sdk/integrations/asyncio.py

@@ -11,7 +11,7 @@
     reraise,
     is_internal_task,
 )
-from sentry_sdk.transport import ASYNC_TRANSPORT_ENABLED, AsyncHttpTransport
+from sentry_sdk.transport import AsyncHttpTransport
 
 try:
     import asyncio
@@ -68,10 +68,7 @@ async def _flush() -> None:
             return
 
         try:
-            if not (
-                ASYNC_TRANSPORT_ENABLED
-                and isinstance(client.transport, AsyncHttpTransport)
-            ):
+            if not isinstance(client.transport, AsyncHttpTransport):
                 return
 
             await client.close_async()