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,26 @@
-import asyncio
-import os
-import weakref
+import threading
 from typing import Dict, Optional
 
 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 (
+    connection_retries,
+    make_async_logging_event_hooks,
+    pool_idle_timeout,
+    pool_max_idle_per_host,
+    proxy_to_url,
+)
+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
-
-limits = Limits(
-    max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS", "20")),
-    max_connections=int(os.getenv("E2B_MAX_CONNECTIONS", "2000")),
-    keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY", "300")),
-)
-
-TransportKey = Optional[ProxyTypes]
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
+def get_api_client(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False, **kwargs
+) -> AsyncVolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -49,42 +47,53 @@ 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": get_transport(config, for_streaming=for_streaming),
             "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, streaming) pair; 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[str], bool], AsyncApiPyqwestTransport] = {}
+
+
+def get_transport(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False
+) -> AsyncApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transports for volume content API calls.
+
+    The streaming transport carries ``read_timeout``, 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. Only streamed downloads use it: reqwest's read timer keeps
+    running while a request body is sent and while waiting for the response
+    head, so on the regular transport it would cut off uploads and slow
+    unary responses longer than the idle bound (those stay bounded by their
+    whole-request deadlines instead).
+    """
+    proxy_url = proxy_to_url(config.proxy)
+    key = (proxy_url, for_streaming)
+    with _transport_lock:
+        transport = _transports.get(key)
+        if transport is None:
+            transport = AsyncApiPyqwestTransport(
+                ConnectionRetryTransport(
+                    HTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=proxy_url,
+                        pool_idle_timeout=pool_idle_timeout,
+                        pool_max_idle_per_host=pool_max_idle_per_host,
+                        read_timeout=READ_TIMEOUT if for_streaming else None,
+                    ),
+                    max_retries=connection_retries,
+                )
+            )
+            _transports[key] = transport
+        return transport

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

@@ -1,27 +1,26 @@
-import os
 import threading
 from typing import Dict, Optional
 
 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 (
+    connection_retries,
+    make_logging_event_hooks,
+    pool_idle_timeout,
+    pool_max_idle_per_host,
+    proxy_to_url,
+)
+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
-
-limits = Limits(
-    max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS", "20")),
-    max_connections=int(os.getenv("E2B_MAX_CONNECTIONS", "2000")),
-    keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY", "300")),
-)
-
-TransportKey = Optional[ProxyTypes]
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
+def get_api_client(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False, **kwargs
+) -> VolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -48,35 +47,52 @@ 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": get_transport(config, for_streaming=for_streaming),
             "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, streaming) pair; 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[str], bool], ApiPyqwestTransport] = {}
 
-    @property
-    def pool(self):
-        return self._pool
 
+def get_transport(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False
+) -> ApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transports 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
+    The streaming transport carries ``read_timeout``, 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. Only streamed downloads use it: reqwest's read timer keeps
+    running while a request body is sent and while waiting for the response
+    head, so on the regular transport it would cut off uploads and slow
+    unary responses longer than the idle bound (those stay bounded by their
+    whole-request deadlines instead).
+    """
+    proxy_url = proxy_to_url(config.proxy)
+    key = (proxy_url, for_streaming)
+    with _transport_lock:
+        transport = _transports.get(key)
+        if transport is None:
+            transport = ApiPyqwestTransport(
+                ConnectionRetryTransport(
+                    SyncHTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=proxy_url,
+                        pool_idle_timeout=pool_idle_timeout,
+                        pool_max_idle_per_host=pool_max_idle_per_host,
+                        read_timeout=READ_TIMEOUT if for_streaming else None,
+                    ),
+                    max_retries=connection_retries,
+                )
+            )
+            _transports[key] = transport
+        return transport

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

@@ -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):
     """