feat(python-sdk): move the volume content client onto pyqwest

.changeset/python-pyqwest-volume-content.md

@@ -0,0 +1,19 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Move the volume content client (`Volume`/`AsyncVolume` file operations) onto
+[`pyqwest`](https://pypi.org/project/pyqwest/) via its httpx-compatible
+transport adapter, the same stack the REST API client uses. The connection
+pool is shared process-wide per proxy instead of one pool per thread (sync)
+or per event loop (async), and connection-establishment failures are retried
+with backoff (`E2B_CONNECTION_RETRIES`, default 3), as before.
+
+For streamed volume reads (`Volume.read_file(format="stream")`), a stalled
+stream is by default bounded by a transport-wide idle read timeout of
+60 seconds that resets on every chunk (still surfaced as
+`httpx.ReadTimeout`; matches the JS SDK's default stream idle timeout).
+`AsyncVolume.read_file` keeps honoring an explicit `stream_idle_timeout`
+per read (including `0` to disable); the sync client ignores it — it cannot
+interrupt a blocking read. Passing `request_timeout` to a streamed read now
+bounds the whole transfer rather than individual socket operations.

packages/python-sdk/e2b/volume/client_async/__init__.py

@@ -1,28 +1,40 @@
-import asyncio
-import os
-import weakref
-from typing import Dict, Optional
+import threading
+from typing import Dict, Optional, Tuple
 
 import httpx
-from httpx import Limits
-from httpx._types import ProxyTypes
-
-from e2b.api import connection_retries, make_async_logging_event_hooks
+from pyqwest import HTTPTransport
+
+from e2b.api import (
+    ProxyConfig,
+    connection_retries,
+    make_async_logging_event_hooks,
+    pool_idle_timeout,
+    pool_max_idle_per_host,
+    proxy_to_config,
+)
+from e2b.api.client_async import AsyncApiPyqwestTransport, ConnectionRetryTransport
 from e2b.api.metadata import default_headers
 from e2b.exceptions import AuthenticationException
 from e2b.volume.client.client import AuthenticatedClient as AsyncVolumeApiClient
-from e2b.volume.connection_config import VolumeConnectionConfig
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
-limits = Limits(
-    max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20"),
-    max_connections=int(os.getenv("E2B_MAX_CONNECTIONS") or "2000"),
-    keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300"),
-)
 
-TransportKey = Optional[ProxyTypes]
+def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
+    """The client for volume content API calls."""
+    return _api_client(config, get_transport(config), **kwargs)
 
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
+def get_streaming_api_client(
+    config: VolumeConnectionConfig, **kwargs
+) -> AsyncVolumeApiClient:
+    """The client for streamed downloads: the same client on the streaming
+    transport, which bounds a stalled read (see :func:`get_streaming_transport`)."""
+    return _api_client(config, get_streaming_transport(config), **kwargs)
+
+
+def _api_client(
+    config: VolumeConnectionConfig, transport: AsyncApiPyqwestTransport, **kwargs
+) -> AsyncVolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -49,42 +61,67 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiCl
         httpx_args={
             # The proxy lives in the cached transport; passing `proxy` here too
             # would mount a fresh, never-closed proxy transport per client.
-            "transport": get_transport(config),
+            "transport": transport,
             "event_hooks": make_async_logging_event_hooks(config.logger),
         },
         **kwargs,
     )
 
 
-class AsyncTransportWithLogger(httpx.AsyncHTTPTransport):
-    # Keyed weakly by the event loop object itself, not id(loop) — CPython
-    # reuses object ids, so a new loop could otherwise inherit a transport
-    # bound to a previous, closed loop.
-    _instances: weakref.WeakKeyDictionary[
-        asyncio.AbstractEventLoop,
-        Dict[TransportKey, "AsyncTransportWithLogger"],
-    ] = weakref.WeakKeyDictionary()
-
-    @property
-    def pool(self):
-        return self._pool
-
-
-def get_transport(config: VolumeConnectionConfig) -> AsyncTransportWithLogger:
-    loop = asyncio.get_running_loop()
-    loop_instances = AsyncTransportWithLogger._instances.get(loop)
-    if loop_instances is None:
-        loop_instances = {}
-        AsyncTransportWithLogger._instances[loop] = loop_instances
-
-    key: TransportKey = config.proxy
-    transport = loop_instances.get(key)
-    if transport is None:
-        transport = AsyncTransportWithLogger(
-            limits=limits,
-            proxy=config.proxy,
-            retries=connection_retries,
-        )
-        loop_instances[key] = transport
-
-    return transport
+_transport_lock = threading.Lock()
+# One transport (= one connection pool) per proxy and read timeout; None is
+# the direct pool. pyqwest's I/O runs on its own Rust runtime, so unlike the
+# httpx transport this replaced, the transport is not bound to an event loop
+# and the cache is process-global rather than per-loop.
+_transports: Dict[
+    Tuple[Optional[ProxyConfig], Optional[float]], AsyncApiPyqwestTransport
+] = {}
+
+
+def get_transport(config: VolumeConnectionConfig) -> AsyncApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transport for volume content API calls.
+
+    It carries no idle read bound: reqwest's read timer keeps running while a
+    request body is sent and while waiting for the response head, so one here
+    would cut off uploads and slow unary responses (they stay bounded by
+    their whole-request deadlines instead). Streamed downloads, which do need
+    an idle bound, use :func:`get_streaming_transport`.
+    """
+    return _transport(config, read_timeout=None)
+
+
+def get_streaming_transport(
+    config: VolumeConnectionConfig,
+) -> AsyncApiPyqwestTransport:
+    """The transport for streamed downloads, carrying ``READ_TIMEOUT`` as the
+    idle bound on every read: it resets after each successful read, so it caps
+    how long a streamed download may stall without limiting total transfer
+    time. It is fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines, and the sync adapter does not bound body reads at
+    all.
+    """
+    return _transport(config, read_timeout=READ_TIMEOUT)
+
+
+def _transport(
+    config: VolumeConnectionConfig, *, read_timeout: Optional[float]
+) -> AsyncApiPyqwestTransport:
+    proxy = proxy_to_config(config.proxy)
+    key = (proxy, read_timeout)
+    with _transport_lock:
+        transport = _transports.get(key)
+        if transport is None:
+            transport = AsyncApiPyqwestTransport(
+                ConnectionRetryTransport(
+                    HTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=proxy.to_pyqwest() if proxy is not None else None,
+                        pool_idle_timeout=pool_idle_timeout,
+                        pool_max_idle_per_host=pool_max_idle_per_host,
+                        read_timeout=read_timeout,
+                    ),
+                    max_retries=connection_retries,
+                )
+            )
+            _transports[key] = transport
+        return transport

packages/python-sdk/e2b/volume/client_sync/__init__.py

@@ -1,27 +1,40 @@
-import os
 import threading
-from typing import Dict, Optional
+from typing import Dict, Optional, Tuple
 
 import httpx
-from httpx import Limits
-from httpx._types import ProxyTypes
+from pyqwest import SyncHTTPTransport
 
-from e2b.api import connection_retries, make_logging_event_hooks
+from e2b.api import (
+    ProxyConfig,
+    connection_retries,
+    make_logging_event_hooks,
+    pool_idle_timeout,
+    pool_max_idle_per_host,
+    proxy_to_config,
+)
+from e2b.api.client_sync import ApiPyqwestTransport, ConnectionRetryTransport
 from e2b.api.metadata import default_headers
 from e2b.exceptions import AuthenticationException
 from e2b.volume.client.client import AuthenticatedClient as VolumeApiClient
-from e2b.volume.connection_config import VolumeConnectionConfig
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
-limits = Limits(
-    max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20"),
-    max_connections=int(os.getenv("E2B_MAX_CONNECTIONS") or "2000"),
-    keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300"),
-)
 
-TransportKey = Optional[ProxyTypes]
+def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
+    """The client for volume content API calls."""
+    return _api_client(config, get_transport(config), **kwargs)
+
 
+def get_streaming_api_client(
+    config: VolumeConnectionConfig, **kwargs
+) -> VolumeApiClient:
+    """The client for streamed downloads: the same client on the streaming
+    transport, which bounds a stalled read (see :func:`get_streaming_transport`)."""
+    return _api_client(config, get_streaming_transport(config), **kwargs)
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
+
+def _api_client(
+    config: VolumeConnectionConfig, transport: ApiPyqwestTransport, **kwargs
+) -> VolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -48,35 +61,64 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
         httpx_args={
             # The proxy lives in the cached transport; passing `proxy` here too
             # would mount a fresh, never-closed proxy transport per client.
-            "transport": get_transport(config),
+            "transport": transport,
             "event_hooks": make_logging_event_hooks(config.logger),
         },
         **kwargs,
     )
 
 
-class TransportWithLogger(httpx.HTTPTransport):
-    _thread_local = threading.local()
+_transport_lock = threading.Lock()
+# One transport (= one connection pool) per proxy and read timeout; None is
+# the direct pool. pyqwest transports are thread-safe, so unlike the httpx
+# transport this replaced, the cache is process-global rather than per-thread.
+_transports: Dict[
+    Tuple[Optional[ProxyConfig], Optional[float]], ApiPyqwestTransport
+] = {}
 
-    @property
-    def pool(self):
-        return self._pool
 
+def get_transport(config: VolumeConnectionConfig) -> ApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transport for volume content API calls.
 
-def get_transport(config: VolumeConnectionConfig) -> TransportWithLogger:
-    instances: Dict[TransportKey, TransportWithLogger] = getattr(
-        TransportWithLogger._thread_local, "instances", {}
-    )
-    key: TransportKey = config.proxy
-    cached = instances.get(key)
-    if cached is not None:
-        return cached
-
-    transport = TransportWithLogger(
-        limits=limits,
-        proxy=config.proxy,
-        retries=connection_retries,
-    )
-    instances[key] = transport
-    TransportWithLogger._thread_local.instances = instances
-    return transport
+    It carries no idle read bound: reqwest's read timer keeps running while a
+    request body is sent and while waiting for the response head, so one here
+    would cut off uploads and slow unary responses (they stay bounded by
+    their whole-request deadlines instead). Streamed downloads, which do need
+    an idle bound, use :func:`get_streaming_transport`.
+    """
+    return _transport(config, read_timeout=None)
+
+
+def get_streaming_transport(config: VolumeConnectionConfig) -> ApiPyqwestTransport:
+    """The transport for streamed downloads, carrying ``READ_TIMEOUT`` as the
+    idle bound on every read: it resets after each successful read, so it caps
+    how long a streamed download may stall without limiting total transfer
+    time. It is fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines, and the sync adapter does not bound body reads at
+    all.
+    """
+    return _transport(config, read_timeout=READ_TIMEOUT)
+
+
+def _transport(
+    config: VolumeConnectionConfig, *, read_timeout: Optional[float]
+) -> ApiPyqwestTransport:
+    proxy = proxy_to_config(config.proxy)
+    key = (proxy, read_timeout)
+    with _transport_lock:
+        transport = _transports.get(key)
+        if transport is None:
+            transport = ApiPyqwestTransport(
+                ConnectionRetryTransport(
+                    SyncHTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=proxy.to_pyqwest() if proxy is not None else None,
+                        pool_idle_timeout=pool_idle_timeout,
+                        pool_max_idle_per_host=pool_max_idle_per_host,
+                        read_timeout=read_timeout,
+                    ),
+                    max_retries=connection_retries,
+                )
+            )
+            _transports[key] = transport
+        return transport

packages/python-sdk/e2b/volume/connection_config.py

@@ -3,10 +3,10 @@
 
 from typing import Dict, Optional, TypedDict
 
-from httpx._types import ProxyTypes
 from typing_extensions import Unpack
 
 from e2b.api.metadata import package_version
+from e2b.connection_config import ProxyTypes
 
 REQUEST_TIMEOUT: float = 60.0  # 60 seconds
 
@@ -15,6 +15,12 @@
 # bounds each chunk by the request timeout and leaves the total to the server.)
 FILE_TIMEOUT: float = 3600.0  # 1 hour
 
+# Idle bound for every read on the volume content transports: the transfer is
+# aborted when no bytes at all arrive for this long. It resets on each chunk,
+# so it never limits total transfer time — only a fully stalled connection.
+# Matches the JS SDK's default stream idle timeout (REQUEST_TIMEOUT_MS).
+READ_TIMEOUT: float = 60.0  # 60 seconds
+
 
 class VolumeApiParams(TypedDict, total=False):
     """