Skip to content

Commit e2d6a0c

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 fe6b157 commit e2d6a0c

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-
from httpx._types import ProxyTypes
9-
10-
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
1116
from e2b.api.metadata import default_headers
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
7-
from httpx._types import ProxyTypes
5+
from pyqwest import SyncHTTPTransport
86

9-
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
1016
from e2b.api.metadata import default_headers
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

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

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

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

518525
return stream_file()
519526

0 commit comments

Comments
 (0)