Skip to content

Commit 85a736f

Browse files
mishushakovclaude
mishushakov
and
claude
committed
feat(python-sdk): move volume content client onto pyqwest
Volume/AsyncVolume file operations join the API client on the ApiPyqwestTransport + connection-retry stack; transport caches become process-global keyed by proxy. httpx.Proxy values reduce to plain proxy URLs; stream_idle_timeout is accepted but a no-op at this point (bounded by the transport in the next commits). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 3c94acf commit 85a736f

6 files changed

Lines changed: 335 additions & 211 deletions

File tree

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

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

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

Lines changed: 48 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -1,25 +1,22 @@
1-
import asyncio
2-
import os
3-
import weakref
1+
import threading
42
from typing import Dict, Optional
53

64
import httpx
7-
from httpx import Limits
8-
9-
from e2b.api import connection_retries, make_async_logging_event_hooks
5+
from pyqwest import HTTPTransport
6+
7+
from e2b.api import (
8+
ProxyConfig,
9+
connection_retries,
10+
make_async_logging_event_hooks,
11+
pool_idle_timeout,
12+
pool_max_idle_per_host,
13+
proxy_to_config,
14+
)
15+
from e2b.api.client_async import AsyncApiPyqwestTransport, ConnectionRetryTransport
1016
from e2b.api.metadata import default_headers
11-
from e2b.connection_config import ProxyTypes
1217
from e2b.exceptions import AuthenticationException
1318
from e2b.volume.client.client import AuthenticatedClient as AsyncVolumeApiClient
14-
from e2b.volume.connection_config import VolumeConnectionConfig
15-
16-
limits = Limits(
17-
max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20"),
18-
max_connections=int(os.getenv("E2B_MAX_CONNECTIONS") or "2000"),
19-
keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300"),
20-
)
21-
22-
TransportKey = Optional[ProxyTypes]
19+
from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
2320

2421

2522
def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
@@ -56,35 +53,38 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiCl
5653
)
5754

5855

59-
class AsyncTransportWithLogger(httpx.AsyncHTTPTransport):
60-
# Keyed weakly by the event loop object itself, not id(loop) — CPython
61-
# reuses object ids, so a new loop could otherwise inherit a transport
62-
# bound to a previous, closed loop.
63-
_instances: weakref.WeakKeyDictionary[
64-
asyncio.AbstractEventLoop,
65-
Dict[TransportKey, "AsyncTransportWithLogger"],
66-
] = weakref.WeakKeyDictionary()
67-
68-
@property
69-
def pool(self):
70-
return self._pool
71-
72-
73-
def get_transport(config: VolumeConnectionConfig) -> AsyncTransportWithLogger:
74-
loop = asyncio.get_running_loop()
75-
loop_instances = AsyncTransportWithLogger._instances.get(loop)
76-
if loop_instances is None:
77-
loop_instances = {}
78-
AsyncTransportWithLogger._instances[loop] = loop_instances
79-
80-
key: TransportKey = config.proxy
81-
transport = loop_instances.get(key)
82-
if transport is None:
83-
transport = AsyncTransportWithLogger(
84-
limits=limits,
85-
proxy=config.proxy,
86-
retries=connection_retries,
87-
)
88-
loop_instances[key] = transport
89-
90-
return transport
56+
_transport_lock = threading.Lock()
57+
# One transport (= one connection pool) per proxy; None is the direct pool.
58+
# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx transport
59+
# this replaced, the transport is not bound to an event loop and the cache
60+
# is process-global rather than per-loop.
61+
_transports: Dict[Optional[ProxyConfig], AsyncApiPyqwestTransport] = {}
62+
63+
64+
def get_transport(config: VolumeConnectionConfig) -> AsyncApiPyqwestTransport:
65+
"""The shared pyqwest-backed httpx transport for volume content API calls.
66+
67+
``read_timeout`` is the idle bound on every read: it resets after each
68+
successful read, so it caps how long a streamed download may stall
69+
without limiting total transfer time. It is fixed per transport — the
70+
adapter's per-request timeouts are whole-request deadlines, and the sync
71+
adapter does not bound body reads at all.
72+
"""
73+
proxy = proxy_to_config(config.proxy)
74+
with _transport_lock:
75+
transport = _transports.get(proxy)
76+
if transport is None:
77+
transport = AsyncApiPyqwestTransport(
78+
ConnectionRetryTransport(
79+
HTTPTransport(
80+
tls_include_system_certs=True,
81+
proxy=proxy.to_pyqwest() if proxy is not None else None,
82+
pool_idle_timeout=pool_idle_timeout,
83+
pool_max_idle_per_host=pool_max_idle_per_host,
84+
read_timeout=FILE_TIMEOUT,
85+
),
86+
max_retries=connection_retries,
87+
)
88+
)
89+
_transports[proxy] = transport
90+
return transport

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

Lines changed: 42 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,22 @@
1-
import os
21
import threading
32
from typing import Dict, Optional
43

54
import httpx
6-
from httpx import Limits
5+
from pyqwest import SyncHTTPTransport
76

8-
from e2b.api import connection_retries, make_logging_event_hooks
7+
from e2b.api import (
8+
ProxyConfig,
9+
connection_retries,
10+
make_logging_event_hooks,
11+
pool_idle_timeout,
12+
pool_max_idle_per_host,
13+
proxy_to_config,
14+
)
15+
from e2b.api.client_sync import ApiPyqwestTransport, ConnectionRetryTransport
916
from e2b.api.metadata import default_headers
10-
from e2b.connection_config import ProxyTypes
1117
from e2b.exceptions import AuthenticationException
1218
from e2b.volume.client.client import AuthenticatedClient as VolumeApiClient
13-
from e2b.volume.connection_config import VolumeConnectionConfig
14-
15-
limits = Limits(
16-
max_keepalive_connections=int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20"),
17-
max_connections=int(os.getenv("E2B_MAX_CONNECTIONS") or "2000"),
18-
keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300"),
19-
)
20-
21-
TransportKey = Optional[ProxyTypes]
19+
from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
2220

2321

2422
def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
@@ -55,28 +53,37 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
5553
)
5654

5755

58-
class TransportWithLogger(httpx.HTTPTransport):
59-
_thread_local = threading.local()
56+
_transport_lock = threading.Lock()
57+
# One transport (= one connection pool) per proxy; None is the direct pool.
58+
# pyqwest transports are thread-safe, so unlike the httpx transport this
59+
# replaced, the cache is process-global rather than per-thread.
60+
_transports: Dict[Optional[ProxyConfig], ApiPyqwestTransport] = {}
6061

61-
@property
62-
def pool(self):
63-
return self._pool
6462

63+
def get_transport(config: VolumeConnectionConfig) -> ApiPyqwestTransport:
64+
"""The shared pyqwest-backed httpx transport for volume content API calls.
6565
66-
def get_transport(config: VolumeConnectionConfig) -> TransportWithLogger:
67-
instances: Dict[TransportKey, TransportWithLogger] = getattr(
68-
TransportWithLogger._thread_local, "instances", {}
69-
)
70-
key: TransportKey = config.proxy
71-
cached = instances.get(key)
72-
if cached is not None:
73-
return cached
74-
75-
transport = TransportWithLogger(
76-
limits=limits,
77-
proxy=config.proxy,
78-
retries=connection_retries,
79-
)
80-
instances[key] = transport
81-
TransportWithLogger._thread_local.instances = instances
82-
return transport
66+
``read_timeout`` is the idle bound on every read: it resets after each
67+
successful read, so it caps how long a streamed download may stall
68+
without limiting total transfer time. It is fixed per transport — the
69+
adapter's per-request timeouts are whole-request deadlines, and the sync
70+
adapter does not bound body reads at all.
71+
"""
72+
proxy = proxy_to_config(config.proxy)
73+
with _transport_lock:
74+
transport = _transports.get(proxy)
75+
if transport is None:
76+
transport = ApiPyqwestTransport(
77+
ConnectionRetryTransport(
78+
SyncHTTPTransport(
79+
tls_include_system_certs=True,
80+
proxy=proxy.to_pyqwest() if proxy is not None else None,
81+
pool_idle_timeout=pool_idle_timeout,
82+
pool_max_idle_per_host=pool_max_idle_per_host,
83+
read_timeout=FILE_TIMEOUT,
84+
),
85+
max_retries=connection_retries,
86+
)
87+
)
88+
_transports[proxy] = transport
89+
return transport

‎packages/python-sdk/e2b/volume/volume_async.py‎

Lines changed: 41 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import asyncio
2+
13
from typing import AsyncIterator, IO, List, Literal, Optional, Union, cast, overload
24
from http import HTTPStatus
35

@@ -463,11 +465,9 @@ async def read_file(
463465
464466
:param path: Path to the file
465467
:param format: Format of the file content—`text` by default
466-
:param stream_idle_timeout: Idle timeout in **seconds** for a streamed
467-
read (`format="stream"`)—abort if no chunk arrives within this
468-
window while reading. Resets on every chunk, so it bounds a stalled
469-
stream without limiting total transfer time. Defaults to the request
470-
timeout; pass `0` to disable.
468+
:param stream_idle_timeout: Deprecated and ignored. A stalled streamed
469+
read is bounded by a transport-wide idle read timeout instead,
470+
which resets on every chunk.
471471
:param opts: Connection options
472472
473473
:return: File content as string, bytes, or async iterator of bytes
@@ -481,38 +481,45 @@ async def read_file(
481481
)
482482

483483
if format == "stream":
484-
# The request timeout bounds connection setup, not total transfer;
485-
# consuming the body must not be killed by it. httpx's per-chunk
486-
# `read` timeout becomes the idle-read timeout for the body
487-
# (defaults to the request timeout), bounding a stalled stream
488-
# without limiting total transfer time. Pass `0` to disable.
489-
# Mirrors the sandbox files stream path.
490-
idle_timeout = (
491-
timeout if stream_idle_timeout is None else stream_idle_timeout
484+
# Through the pyqwest adapter a per-request timeout is a
485+
# whole-request deadline that would kill long downloads, so a
486+
# streamed read is sent with one only when the caller set
487+
# `request_timeout` explicitly (making it the total-transfer
488+
# deadline). A stalled stream is instead bounded by the
489+
# transport-wide idle read timeout (see `get_transport`), which
490+
# resets on every chunk without limiting total transfer time.
491+
stream_timeout = VolumeConnectionConfig._get_request_timeout(
492+
None, opts.get("request_timeout")
492493
)
493-
stream_timeout = httpx.Timeout(timeout, read=idle_timeout or None)
494494

495495
async def stream_file() -> AsyncIterator[bytes]:
496-
async with api_client.get_async_httpx_client().stream(
497-
method="GET",
498-
url=f"/volumecontent/{self._volume_id}/file",
499-
params=params,
500-
timeout=stream_timeout,
501-
) as response:
502-
if response.status_code == 404:
503-
raise NotFoundException(f"Path {path} not found")
504-
505-
if response.status_code >= 300:
506-
api_response = Response(
507-
status_code=HTTPStatus(response.status_code),
508-
content=await response.aread(),
509-
headers=response.headers,
510-
parsed=None,
511-
)
512-
raise handle_api_exception(api_response, VolumeException)
513-
514-
async for chunk in response.aiter_bytes():
515-
yield chunk
496+
# pyqwest raises the builtin TimeoutError; keep the httpx
497+
# exception the streamed-read contract established.
498+
try:
499+
async with api_client.get_async_httpx_client().stream(
500+
method="GET",
501+
url=f"/volumecontent/{self._volume_id}/file",
502+
params=params,
503+
timeout=stream_timeout,
504+
) as response:
505+
if response.status_code == 404:
506+
raise NotFoundException(f"Path {path} not found")
507+
508+
if response.status_code >= 300:
509+
api_response = Response(
510+
status_code=HTTPStatus(response.status_code),
511+
content=await response.aread(),
512+
headers=response.headers,
513+
parsed=None,
514+
)
515+
raise handle_api_exception(api_response, VolumeException)
516+
517+
async for chunk in response.aiter_bytes():
518+
yield chunk
519+
# asyncio.TimeoutError is distinct from the builtin until 3.11,
520+
# and the adapter's whole-request deadline raises it.
521+
except (TimeoutError, asyncio.TimeoutError) as e:
522+
raise httpx.ReadTimeout(str(e)) from e
516523

517524
return stream_file()
518525

0 commit comments

Comments
 (0)