feat(python-sdk): move template build-context uploads onto pyqwest

.changeset/python-pyqwest-template-uploads.md

@@ -0,0 +1,11 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Move template build-context uploads (to S3 presigned URLs) onto
+[`pyqwest`](https://pypi.org/project/pyqwest/) via its httpx-compatible
+transport adapter. Content-Length framing for the streamed archive body is
+preserved (S3 rejects chunked transfer encoding). The 1-hour upload timeout
+now bounds the entire upload rather than each socket operation, and
+`verify_ssl=False` on the client is no longer honored for uploads (pyqwest
+has no insecure-TLS option).

.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/template_async/build_api.py

@@ -4,8 +4,10 @@
 from typing import Callable, Optional, List, Union
 
 import httpx
+from pyqwest import HTTPTransport
 
-from e2b.api import handle_api_exception
+from e2b.api import handle_api_exception, proxy_to_config
+from e2b.api.client_async import AsyncApiPyqwestTransport
 from e2b.io_utils import aiter_io_chunks
 from e2b.api.client.api.templates import (
     post_v3_templates,
@@ -118,23 +120,35 @@ async def upload_file(
     upload_timeout = (
         request_timeout if request_timeout is not None else FILE_UPLOAD_TIMEOUT_SECONDS
     )
+    upload_proxy = proxy_to_config(getattr(api_client, "_proxy", None))
     try:
         tar_file = tar_file_stream(
             file_name, context_path, ignore_patterns, resolve_symlinks, gzip
         )
         try:
             size = os.fstat(tar_file.fileno()).st_size
 
+            # Through the pyqwest adapter the upload timeout is a
+            # whole-request deadline for the entire transfer, not a per-write
+            # bound as with the httpx transport this replaced.
             async with httpx.AsyncClient(
                 timeout=httpx.Timeout(upload_timeout),
-                verify=api_client._verify_ssl,
                 follow_redirects=api_client._follow_redirects,
-                proxy=getattr(api_client, "_proxy", None),
-                http2=False,
+                transport=AsyncApiPyqwestTransport(
+                    HTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=(
+                            upload_proxy.to_pyqwest()
+                            if upload_proxy is not None
+                            else None
+                        ),
+                    )
+                ),
             ) as client:
                 # Stream the archive from disk via an async iterator. The
                 # explicit Content-Length suppresses chunked transfer
-                # encoding, which S3 presigned URLs reject.
+                # encoding, which S3 presigned URLs reject; reqwest keeps the
+                # Content-Length framing for the streamed body.
                 response = await client.put(
                     url,
                     content=aiter_io_chunks(tar_file),

packages/python-sdk/e2b/template_sync/build_api.py

@@ -3,8 +3,10 @@
 from typing import Callable, Optional, List, Union
 
 import httpx
+from pyqwest import SyncHTTPTransport
 
-from e2b.api import handle_api_exception
+from e2b.api import handle_api_exception, proxy_to_config
+from e2b.api.client_sync import ApiPyqwestTransport
 from e2b.api.client.api.templates import (
     post_v3_templates,
     get_templates_template_id_files_hash,
@@ -116,21 +118,33 @@ def upload_file(
     upload_timeout = (
         request_timeout if request_timeout is not None else FILE_UPLOAD_TIMEOUT_SECONDS
     )
+    upload_proxy = proxy_to_config(getattr(api_client, "_proxy", None))
     try:
         tar_file = tar_file_stream(
             file_name, context_path, ignore_patterns, resolve_symlinks, gzip
         )
         try:
+            # Through the pyqwest adapter the upload timeout is a
+            # whole-request deadline for the entire transfer, not a per-write
+            # bound as with the httpx transport this replaced.
             with httpx.Client(
                 timeout=httpx.Timeout(upload_timeout),
-                verify=api_client._verify_ssl,
                 follow_redirects=api_client._follow_redirects,
-                proxy=getattr(api_client, "_proxy", None),
-                http2=False,
+                transport=ApiPyqwestTransport(
+                    SyncHTTPTransport(
+                        tls_include_system_certs=True,
+                        proxy=(
+                            upload_proxy.to_pyqwest()
+                            if upload_proxy is not None
+                            else None
+                        ),
+                    )
+                ),
             ) as client:
                 # httpx streams the archive from disk in chunks and sets
                 # Content-Length from the file size—S3 presigned URLs reject
-                # chunked transfer encoding.
+                # chunked transfer encoding, and reqwest keeps the
+                # Content-Length framing for the streamed body.
                 response = client.put(url, content=tar_file)
             response.raise_for_status()
         finally:

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

@@ -1,28 +1,27 @@
-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 (
+    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
-
-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]
+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 +48,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[ProxyConfig], 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 = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
+    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 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,27 @@
-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 (
+    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
-
-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]
+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 +48,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[ProxyConfig], 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 = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
+    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 if for_streaming else None,
+                    ),
+                    max_retries=connection_retries,
+                )
+            )
+            _transports[key] = transport
+        return transport