Skip to content

Commit 7459367

Browse files
mishushakovclaude
mishushakov
and
claude
committed
feat(python-sdk): move volume content client and template uploads onto pyqwest
The volume content transports move to the shared pyqwest-backed httpx adapter with a transport-wide idle read timeout (verified to reset per read and to cover body reads in both sync and async — the only stall bound the sync adapter offers), replacing the per-call stream_idle_timeout parameter. Streamed reads are sent without a per-request timeout unless the caller passes request_timeout, which the adapter turns into a whole-transfer deadline; pyqwest's builtin TimeoutError is remapped to httpx.ReadTimeout to keep the streamed-read contract. Template build-context uploads switch to the same transports: httpx still derives Content-Length from the spooled archive (sync) or sends it explicitly (async), and reqwest keeps the Content-Length framing for streamed bodies, which S3 presigned URLs require. The upload tests now compare header names case-insensitively since hyper lowercases them where httpcore title-cased. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent fc1dd98 commit 7459367

10 files changed

Lines changed: 353 additions & 236 deletions

File tree

‎.changeset/python-pyqwest-rest-transport.md‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,20 @@
33
---
44

55
Move the REST API client (sandbox lifecycle, listing, templates, volumes
6-
control plane) onto [`pyqwest`](https://pypi.org/project/pyqwest/) (Rust
6+
control plane), the volume content client, and the template build-context
7+
uploads onto [`pyqwest`](https://pypi.org/project/pyqwest/) (Rust
78
reqwest/hyper) via its httpx-compatible transport adapter, replacing the
89
httpx-native `HTTPTransport`/`AsyncHTTPTransport`. The generated httpx client
910
API is unchanged — only the transport underneath is swapped — so logging
1011
event hooks, per-request timeouts, and headers behave as before.
1112

13+
For streamed volume reads (`Volume.read_file(format="stream")`), a stalled
14+
stream is now bounded by a transport-wide idle read timeout that resets on
15+
every chunk (still surfaced as `httpx.ReadTimeout`); the per-call
16+
`stream_idle_timeout` parameter is removed. Passing `request_timeout` to a
17+
streamed read (and to template uploads) now bounds the whole transfer rather
18+
than individual socket operations.
19+
1220
Because pyqwest transports are thread-safe and loop-independent (I/O runs on
1321
a Rust runtime), the API connection pool is now shared process-wide per
1422
proxy, instead of one pool per thread (sync) or per event loop (async), and

‎packages/python-sdk/e2b/template_async/build_api.py‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,10 @@
44
from typing import Callable, Optional, List, Union
55

66
import httpx
7+
from pyqwest import HTTPTransport
78

8-
from e2b.api import handle_api_exception
9+
from e2b.api import handle_api_exception, proxy_to_url
10+
from e2b.api.client_async import AsyncApiPyqwestTransport
911
from e2b.io_utils import aiter_io_chunks
1012
from e2b.api.client.api.templates import (
1113
post_v3_templates,
@@ -125,16 +127,23 @@ async def upload_file(
125127
try:
126128
size = os.fstat(tar_file.fileno()).st_size
127129

130+
# Through the pyqwest adapter the upload timeout is a
131+
# whole-request deadline for the entire transfer, not a per-write
132+
# bound as with the httpx transport this replaced.
128133
async with httpx.AsyncClient(
129134
timeout=httpx.Timeout(upload_timeout),
130-
verify=api_client._verify_ssl,
131135
follow_redirects=api_client._follow_redirects,
132-
proxy=getattr(api_client, "_proxy", None),
133-
http2=False,
136+
transport=AsyncApiPyqwestTransport(
137+
HTTPTransport(
138+
tls_include_system_certs=True,
139+
proxy=proxy_to_url(getattr(api_client, "_proxy", None)),
140+
)
141+
),
134142
) as client:
135143
# Stream the archive from disk via an async iterator. The
136144
# explicit Content-Length suppresses chunked transfer
137-
# encoding, which S3 presigned URLs reject.
145+
# encoding, which S3 presigned URLs reject; reqwest keeps the
146+
# Content-Length framing for the streamed body.
138147
response = await client.put(
139148
url,
140149
content=aiter_io_chunks(tar_file),

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

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,10 @@
33
from typing import Callable, Optional, List, Union
44

55
import httpx
6+
from pyqwest import SyncHTTPTransport
67

7-
from e2b.api import handle_api_exception
8+
from e2b.api import handle_api_exception, proxy_to_url
9+
from e2b.api.client_sync import ApiPyqwestTransport
810
from e2b.api.client.api.templates import (
911
post_v3_templates,
1012
get_templates_template_id_files_hash,
@@ -121,16 +123,23 @@ def upload_file(
121123
file_name, context_path, ignore_patterns, resolve_symlinks, gzip
122124
)
123125
try:
126+
# Through the pyqwest adapter the upload timeout is a
127+
# whole-request deadline for the entire transfer, not a per-write
128+
# bound as with the httpx transport this replaced.
124129
with httpx.Client(
125130
timeout=httpx.Timeout(upload_timeout),
126-
verify=api_client._verify_ssl,
127131
follow_redirects=api_client._follow_redirects,
128-
proxy=getattr(api_client, "_proxy", None),
129-
http2=False,
132+
transport=ApiPyqwestTransport(
133+
SyncHTTPTransport(
134+
tls_include_system_certs=True,
135+
proxy=proxy_to_url(getattr(api_client, "_proxy", None)),
136+
)
137+
),
130138
) as client:
131139
# httpx streams the archive from disk in chunks and sets
132140
# Content-Length from the file size—S3 presigned URLs reject
133-
# chunked transfer encoding.
141+
# chunked transfer encoding, and reqwest keeps the
142+
# Content-Length framing for the streamed body.
134143
response = client.put(url, content=tar_file)
135144
response.raise_for_status()
136145
finally:

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

Lines changed: 47 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -1,25 +1,21 @@
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+
connection_retries,
9+
make_async_logging_event_hooks,
10+
pool_idle_timeout,
11+
pool_max_idle_per_host,
12+
proxy_to_url,
13+
)
14+
from e2b.api.client_async import AsyncApiPyqwestTransport, ConnectionRetryTransport
1115
from e2b.api.metadata import default_headers
1216
from e2b.exceptions import AuthenticationException
1317
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", "20")),
18-
max_connections=int(os.getenv("E2B_MAX_CONNECTIONS", "2000")),
19-
keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY", "300")),
20-
)
21-
22-
TransportKey = Optional[ProxyTypes]
18+
from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
2319

2420

2521
def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
@@ -56,35 +52,38 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiCl
5652
)
5753

5854

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
55+
_transport_lock = threading.Lock()
56+
# One transport (= one connection pool) per proxy; None is the direct pool.
57+
# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx transport
58+
# this replaced, the transport is not bound to an event loop and the cache
59+
# is process-global rather than per-loop.
60+
_transports: Dict[Optional[str], AsyncApiPyqwestTransport] = {}
61+
62+
63+
def get_transport(config: VolumeConnectionConfig) -> AsyncApiPyqwestTransport:
64+
"""The shared pyqwest-backed httpx transport for volume content API calls.
65+
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_url = proxy_to_url(config.proxy)
73+
with _transport_lock:
74+
transport = _transports.get(proxy_url)
75+
if transport is None:
76+
transport = AsyncApiPyqwestTransport(
77+
ConnectionRetryTransport(
78+
HTTPTransport(
79+
tls_include_system_certs=True,
80+
proxy=proxy_url,
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_url] = transport
89+
return transport

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

Lines changed: 41 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,21 @@
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+
connection_retries,
9+
make_logging_event_hooks,
10+
pool_idle_timeout,
11+
pool_max_idle_per_host,
12+
proxy_to_url,
13+
)
14+
from e2b.api.client_sync import ApiPyqwestTransport, ConnectionRetryTransport
1015
from e2b.api.metadata import default_headers
1116
from e2b.exceptions import AuthenticationException
1217
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", "20")),
17-
max_connections=int(os.getenv("E2B_MAX_CONNECTIONS", "2000")),
18-
keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY", "300")),
19-
)
20-
21-
TransportKey = Optional[ProxyTypes]
18+
from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
2219

2320

2421
def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
@@ -55,28 +52,37 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
5552
)
5653

5754

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

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

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

0 commit comments

Comments
 (0)