github-featured/e2b-dev--e2b
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(python-sdk): move the envd HTTP API client onto pyqwest (#1623)

Browse Source 
## What

Tracked in [SDK-265](https://linear.app/e2b/issue/SDK-265) (part of the
[SDK-268](https://linear.app/e2b/issue/SDK-268) stack). Stacked on
#1603, at the top of the pyqwest stack (#1601 → #1602 → #1603 → this).
Migrate the envd HTTP API client — sandbox file transfers
(`files.read`/`write`), health checks — from httpx-native transports to
pyqwest via the httpx adapter, and dedupe the transport plumbing that
#1558 (envd RPC) and #1601 (REST) each carried a copy of. With this, all
Python SDK traffic runs on pyqwest: REST control plane (#1601), envd RPC
(#1558, connectrpc), envd HTTP API (this PR); the volume content client
(#1602) and template build uploads (#1603) sit below this one in the
stack.

## How

**Shared plumbing** (first commit): `e2b.api` becomes the canonical home
for the proxy narrowing (`proxy_to_config`, with stack-neutral error
messages), the pool tuning, and the flavor `ConnectionRetryTransport` +
a new `retrying_http_transport(proxy, read_timeout=None)` factory;
`e2b.envd.client_sync/client_async` import them instead of defining
their own (envd RPC behavior unchanged, pools stay separate —
unification is SDK-291).

**envd HTTP API** (second commit):

- `get_envd_transport(config, for_streaming=False)` returns
pyqwest-adapter transports cached per `(proxy, streaming)`;
`get_envd_api(config, base_url, for_streaming=False)` builds the httpx
client with sandbox headers + logging hooks. The per-thread (sync) /
per-loop (async) client caching in
`Filesystem`/`Commands`/`Pty`/`AsyncSandbox` is gone — one shared client
per module, same rationale as the `ApiClient` simplification in #1601.
- **Streamed downloads**: the streaming transport carries a 60s
`read_timeout` — an idle bound that resets on every read, capping stalls
without limiting total transfer time. It gets a dedicated pool because
reqwest's read timer keeps ticking while a request body is sent and
while waiting for the response head, so on the shared transport it would
cut off uploads and slow unary responses. An explicit `request_timeout`
becomes the whole-transfer deadline (adapter semantics) and is sent only
when the caller set one; `stream_idle_timeout` stays honored on the
async client via `wait_for` per read (so values above 60s work and `0`
disables), and is documented as ignored on the sync client, which cannot
interrupt a blocking read. Mirrors #1602's volume design.
- **Uploads**: buffered uploads keep `request_timeout` as a
whole-request deadline; streamed (file-like) uploads carry no
client-side timeout and are bounded server-side (envd's idle read
timeout) — both exactly the JS SDK's behavior (`getSignal` for buffered,
no signal for streams).
- **Multipart**: `files=` uploads go out as httpx's `MultipartStream`,
which implements *both* `SyncByteStream` and `AsyncByteStream`. The
pyqwest 0.7 adapter's sync content conversion matched `AsyncByteStream`
first and raised `TypeError("unreachable")` from inside the body
iterator, surfacing as a `WriteError` mid-request ("http2 error: stream
error sent by user"). Fixed upstream in
[pyqwest#196](https://github.com/curioswitch/pyqwest/pull/196), which
matches the sync case first — so this PR carries no workaround (the
stack requires **pyqwest 0.9**, set in #1601). The regression test
stays, now covering the upstream fix.
- The stream readers map the transport's idle timeout (builtin
`TimeoutError` under pyqwest) to the documented `httpx.ReadTimeout`;
`handle_envd_api_transport_exception`'s health-probe path keeps working
because the adapter maps HTTP/2 stream resets to
`httpx.RemoteProtocolError`.

- **RPC logging**: the `LoggingInterceptor` docstring no longer promises
its own removal. pyqwest does log requests
([pyqwest#197](https://github.com/curioswitch/pyqwest/pull/197)), but on
process-wide `pyqwest`/`pyqwest.access` loggers that can't carry the
per-sandbox `logger` and don't see streamed messages or the Connect
error code of a stream that fails inside a `200 OK` — so the interceptor
stays, with those loggers below it.
[pyqwest#192](https://github.com/curioswitch/pyqwest/pull/192), the
middleware it referenced, was closed in favor of #197.

- **Transports**: rebased onto #1603 on pyqwest 0.9, so the envd HTTP
API transports are the stock `PyqwestTransport`/`AsyncPyqwestTransport`
(the SDK's adapter subclasses are gone as of #1601 — 0.9 strips the
`Host` header and maps timeouts itself) with `follow_redirects=False`
and, for the streaming pool, the transport-wide `read_timeout`.

## Testing

- Unit: envd transport keying (streaming vs regular vs REST pools),
`get_envd_api` wiring (headers, transports), multipart regression
through a local server, stream-reader timeout mapping + per-read idle
bound (`tests/test_file_stream_reader.py`), rewritten client-lifecycle
tests (shared across threads). 236 unit tests green; lint + typecheck
green.
- Integration against production sandboxes: full `files` suites
sync+async (123 tests — these caught the multipart bug), `commands` +
`pty` suites both flavors (57 tests). All green.

## Usage example

No API changes:

```python
sbx = Sandbox.create()
sbx.files.write("hello.txt", "hi")            # multipart/octet-stream over pyqwest
with sbx.files.read("hello.txt", format="stream") as stream:
    for chunk in stream:                       # stalls bounded by 60s idle read timeout
        ...
```

Only visible behavior shift: on the **sync** client, `files.read(...,
format="stream", stream_idle_timeout=...)` is now a documented no-op
(the transport-wide 60s idle bound applies); the async client honors it
as before.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Mish Ushakov
2026-08-10 19:46:38 +02:00
committed by GitHub
parent b3a7c9f44a
commit b048369307
27 changed files with 694 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/python-pyqwest-envd-api.md
+39
View File
@@ -0,0 +1,39 @@
---
"@e2b/python-sdk": minor
---
Move the envd HTTP API client (sandbox file transfers, health checks) onto
[`pyqwest`](https://pypi.org/project/pyqwest/) via its httpx-compatible
transport adapter. envd RPC already runs on pyqwest through `connectrpc`, so
all sandbox traffic now shares one HTTP stack built from the same transport
pieces (with separate connection pools per use).
The per-thread (sync) and per-loop (async) envd httpx clients are gone: the
pyqwest transports are thread-safe and loop-independent, so a single client
per module serves all threads and event loops.
Timeout semantics through the adapter:
- Streamed downloads (`files.read(format="stream")`): a `request_timeout`
set explicitly for the call is the deadline for the whole transfer — by
default the transfer is unbounded in total, as before. A stalled stream is
reclaimed by a 60-second idle read timeout that resets on every chunk.
`stream_idle_timeout` keeps working on the async client (applied per
read); the sync client cannot interrupt a blocking read, so it relies on
the transport-wide idle bound and now ignores the parameter.
- Uploads: a buffered upload is bounded by `request_timeout` as a
whole-request deadline, and a streamed (file-like) upload carries no
client-side timeout (a stalled one is bounded server-side by envd's idle
read timeout) — both matching the JS SDK.
- Non-streamed reads (`files.read()` as text or bytes) and buffered uploads
are bounded by `request_timeout` for the **whole transfer** (default
60 seconds), where the previous transport bounded each socket operation
and left total duration unbounded. Reading or writing a file too large to
transfer inside the deadline now raises `httpx.ReadTimeout` — pass a
larger `request_timeout` (or `0` to disable), or use
`format="stream"`/file-like data, for large transfers.
`E2B_MAX_CONNECTIONS` is no longer read: it configured httpx's global
connection cap, and the last transport that took one is gone (reqwest has no
counterpart — it does not cap concurrent connections). `E2B_KEEPALIVE_EXPIRY`
and `E2B_MAX_KEEPALIVE_CONNECTIONS` keep tuning the pools.
packages/python-sdk/e2b/api/__init__.py
+10 -17
View File
@@ -7,7 +7,7 @@ from types import TracebackType
from typing import NamedTuple, Optional, Protocol, Tuple, Union
import httpx
from httpx import AsyncBaseTransport, BaseTransport, Limits, Timeout
from httpx import AsyncBaseTransport, BaseTransport, Timeout
from pyqwest import Proxy
from e2b.api.client.client import AuthenticatedClient
@@ -61,19 +61,14 @@ def make_async_logging_event_hooks(log: Optional[logging.Logger]) -> dict:
return {"request": [on_request], "response": [on_response]}
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"),
)
connection_retries = int(os.getenv("E2B_CONNECTION_RETRIES") or "3")
# Mirror the httpx pool tuning above with pyqwest's equivalents for the REST
# API transports. `pool_max_idle_per_host` is per host rather than httpx's
# global idle cap, but API traffic goes to a single host, so the values map
# directly; reqwest has no counterpart to `E2B_MAX_CONNECTIONS` (it does not
# cap concurrent connections).
# Pool tuning for the pyqwest transports, shared by the REST API, envd RPC,
# and envd HTTP API stacks. `pool_max_idle_per_host` is per host rather than
# the global idle cap the httpx transports took, which suits both: API traffic
# goes to a single host and each sandbox is its own host. `E2B_MAX_CONNECTIONS`
# has no counterpart left — reqwest does not cap concurrent connections — so it
# is no longer read.
pool_idle_timeout = float(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300")
pool_max_idle_per_host = int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20")
@@ -109,9 +104,7 @@ def proxy_to_config(proxy: Optional[ProxyTypes]) -> Optional[ProxyConfig]:
return ProxyConfig(str(proxy))
if isinstance(proxy, httpx.Proxy):
if proxy.ssl_context is not None:
raise InvalidArgumentException(
"E2B API calls don't support httpx.Proxy ssl_context"
)
raise InvalidArgumentException("httpx.Proxy ssl_context is not supported")
# httpx.Proxy splits userinfo out of the URL into `.auth`; pyqwest
# takes the credentials the same way, so they pass straight through.
return ProxyConfig(
@@ -120,8 +113,8 @@ def proxy_to_config(proxy: Optional[ProxyTypes]) -> Optional[ProxyConfig]:
headers=tuple(proxy.headers.items()),
)
raise InvalidArgumentException(
"E2B API calls support only URL-string, httpx.URL, and httpx.Proxy "
'proxies, e.g. proxy="http://user:pass@localhost:8030"'
"Only URL-string, httpx.URL, and httpx.Proxy proxies are supported, "
'e.g. proxy="http://user:pass@localhost:8030"'
)
packages/python-sdk/e2b/api/client_async/__init__.py
+90 -65
View File
@@ -1,6 +1,4 @@
import asyncio
import threading
import weakref
from typing import Dict, Optional, Tuple, Union
import httpx
@@ -13,14 +11,12 @@ from e2b.api import (
AsyncApiClient,
ProxyConfig,
connection_retries,
limits,
make_async_logging_event_hooks,
pool_idle_timeout,
pool_max_idle_per_host,
proxy_to_config,
)
from e2b.connection_config import ConnectionConfig, ProxyTypes
TransportKey = Tuple[bool, Optional[ProxyTypes]]
from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
@@ -28,12 +24,14 @@ def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
class ConnectionRetryTransport(RetryTransport):
"""Retry only failures establishing the connection, matching the
connect-only ``retries`` of the httpx transport this replaced: pyqwest
raises the builtin ``ConnectionError`` only before the request was
written, so these retries can never replay a request the API may have
received. The retry middleware's default policy would otherwise also
retry I/O errors and 429/5xx responses for idempotent methods."""
"""Retry only failures establishing the connection — shared by the REST
API and envd RPC stacks: pyqwest raises the builtin ``ConnectionError``
only before the request was written, so these retries can never replay a
request the server may have received (a delivered REST call or unary RPC
like ``SendInput``). This matches the connect-only ``retries`` of the
httpx transports this replaced; the retry middleware's default policy
would otherwise also retry I/O errors and 429/5xx responses for
idempotent methods."""
def should_retry_response(
self, request: Request, response: Union[Response, Exception]
@@ -41,77 +39,104 @@ class ConnectionRetryTransport(RetryTransport):
return isinstance(response, ConnectionError)
def retrying_http_transport(
proxy: Optional[ProxyConfig], read_timeout: Optional[float] = None
) -> ConnectionRetryTransport:
"""A fresh pyqwest transport (= its own connection pool) with the SDK's
shared tuning — system CA certs (without which TLS through an
intercepting proxy fails), the httpx-equivalent pool limits, and
connect-only retries. The REST API, envd RPC, and envd HTTP API stacks
each cache their own instances (pool unification is a follow-up).
``read_timeout`` bounds every read on the transport's connections; see
:func:`get_envd_transport` for when that is (and isn't) appropriate.
Requests are logged by pyqwest itself on the ``pyqwest.access`` and
``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
diagnostics httpcore used to provide. The SDK's own ``logger`` option is
separate and sits above this, on the httpx client."""
return 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,
# Redirects belong to the httpx client above (which the generated
# clients leave off), not to reqwest.
follow_redirects=False,
),
max_retries=connection_retries,
)
_transport_lock = threading.Lock()
# One transport (= one connection pool) per proxy; None is the direct pool.
# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx envd
# transports below, the transport is not bound to an event loop and the
# cache is process-global rather than per-loop.
# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx transports
# they replaced, the transports are not bound to an event loop and the
# caches are process-global rather than per-loop.
_transports: Dict[Optional[ProxyConfig], AsyncPyqwestTransport] = {}
def get_transport(config: ConnectionConfig) -> AsyncPyqwestTransport:
"""The shared pyqwest-backed httpx transport for REST API calls. For TLS
connections ALPN negotiates the HTTP version (HTTP/2 against the E2B
API), like the http2-enabled httpx transport this replaced.
Requests are logged by pyqwest itself on the ``pyqwest.access`` and
``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
diagnostics httpcore used to provide. The SDK's own ``logger`` option is
separate and sits above this, on the httpx client."""
API), like the http2-enabled httpx transport this replaced."""
proxy = proxy_to_config(config.proxy)
with _transport_lock:
transport = _transports.get(proxy)
if transport is None:
transport = AsyncPyqwestTransport(
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,
# Redirects belong to the httpx client above (which the
# generated clients leave off), not to reqwest.
follow_redirects=False,
),
max_retries=connection_retries,
)
)
transport = AsyncPyqwestTransport(retrying_http_transport(proxy))
_transports[proxy] = transport
return transport
class AsyncEnvdTransportWithLogger(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, "AsyncEnvdTransportWithLogger"],
] = weakref.WeakKeyDictionary()
@property
def pool(self):
return self._pool
# One transport per (proxy, streaming) pair, separate from the REST API
# pools — envd traffic goes to per-sandbox hosts.
_envd_transports: Dict[Tuple[Optional[ProxyConfig], bool], AsyncPyqwestTransport] = {}
def get_envd_transport(
config: ConnectionConfig, http2: bool = True
) -> AsyncEnvdTransportWithLogger:
loop = asyncio.get_running_loop()
loop_instances = AsyncEnvdTransportWithLogger._instances.get(loop)
if loop_instances is None:
loop_instances = {}
AsyncEnvdTransportWithLogger._instances[loop] = loop_instances
config: ConnectionConfig, *, for_streaming: bool = False
) -> AsyncPyqwestTransport:
"""The shared pyqwest-backed httpx transports for the envd HTTP API
(file transfers, health checks).
key: TransportKey = (http2, config.proxy)
transport = loop_instances.get(key)
if transport is None:
transport = AsyncEnvdTransportWithLogger(
limits=limits,
proxy=config.proxy,
http2=http2,
retries=connection_retries,
)
loop_instances[key] = 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. 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 = _envd_transports.get(key)
if transport is None:
transport = AsyncPyqwestTransport(
retrying_http_transport(
proxy,
read_timeout=READ_TIMEOUT if for_streaming else None,
)
)
_envd_transports[key] = transport
return transport
return transport
def get_envd_api(
config: ConnectionConfig, base_url: str, *, for_streaming: bool = False
) -> httpx.AsyncClient:
"""An httpx client for a sandbox's envd HTTP API (file transfers, health
checks) on the shared pyqwest transports. The client itself is a cheap
stateless wrapper — one per consumer is fine — while the pooled transport
underneath is shared and loop-independent."""
return httpx.AsyncClient(
base_url=base_url,
transport=get_envd_transport(config, for_streaming=for_streaming),
headers=config.sandbox_headers,
event_hooks=make_async_logging_event_hooks(config.logger),
)
packages/python-sdk/e2b/api/client_sync/__init__.py
+90 -55
View File
@@ -11,14 +11,12 @@ from e2b.api import (
ApiClient,
ProxyConfig,
connection_retries,
limits,
make_logging_event_hooks,
pool_idle_timeout,
pool_max_idle_per_host,
proxy_to_config,
)
from e2b.connection_config import ConnectionConfig, ProxyTypes
TransportKey = Tuple[bool, Optional[ProxyTypes]]
from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
def get_api_client(config: ConnectionConfig, **kwargs) -> ApiClient:
@@ -26,12 +24,14 @@ def get_api_client(config: ConnectionConfig, **kwargs) -> ApiClient:
class ConnectionRetryTransport(SyncRetryTransport):
"""Retry only failures establishing the connection, matching the
connect-only ``retries`` of the httpx transport this replaced: pyqwest
raises the builtin ``ConnectionError`` only before the request was
written, so these retries can never replay a request the API may have
received. The retry middleware's default policy would otherwise also
retry I/O errors and 429/5xx responses for idempotent methods."""
"""Retry only failures establishing the connection — shared by the REST
API and envd RPC stacks: pyqwest raises the builtin ``ConnectionError``
only before the request was written, so these retries can never replay a
request the server may have received (a delivered REST call or unary RPC
like ``SendInput``). This matches the connect-only ``retries`` of the
httpx transports this replaced; the retry middleware's default policy
would otherwise also retry I/O errors and 429/5xx responses for
idempotent methods."""
def should_retry_response(
self, request: SyncRequest, response: Union[SyncResponse, Exception]
@@ -39,69 +39,104 @@ class ConnectionRetryTransport(SyncRetryTransport):
return isinstance(response, ConnectionError)
def retrying_http_transport(
proxy: Optional[ProxyConfig], read_timeout: Optional[float] = None
) -> ConnectionRetryTransport:
"""A fresh pyqwest transport (= its own connection pool) with the SDK's
shared tuning — system CA certs (without which TLS through an
intercepting proxy fails), the httpx-equivalent pool limits, and
connect-only retries. The REST API, envd RPC, and envd HTTP API stacks
each cache their own instances (pool unification is a follow-up).
``read_timeout`` bounds every read on the transport's connections; see
:func:`get_envd_transport` for when that is (and isn't) appropriate.
Requests are logged by pyqwest itself on the ``pyqwest.access`` and
``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
diagnostics httpcore used to provide. The SDK's own ``logger`` option is
separate and sits above this, on the httpx client."""
return 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,
# Redirects belong to the httpx client above (which the generated
# clients leave off), not to reqwest.
follow_redirects=False,
),
max_retries=connection_retries,
)
_transport_lock = threading.Lock()
# One transport (= one connection pool) per proxy; None is the direct pool.
# pyqwest transports are thread-safe, so unlike the httpx envd transports
# below, the cache is process-global rather than per-thread.
# pyqwest transports are thread-safe, so unlike the httpx transports they
# replaced, the caches are process-global rather than per-thread.
_transports: Dict[Optional[ProxyConfig], PyqwestTransport] = {}
def get_transport(config: ConnectionConfig) -> PyqwestTransport:
"""The shared pyqwest-backed httpx transport for REST API calls. For TLS
connections ALPN negotiates the HTTP version (HTTP/2 against the E2B
API), like the http2-enabled httpx transport this replaced.
Requests are logged by pyqwest itself on the ``pyqwest.access`` and
``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
diagnostics httpcore used to provide. The SDK's own ``logger`` option is
separate and sits above this, on the httpx client."""
API), like the http2-enabled httpx transport this replaced."""
proxy = proxy_to_config(config.proxy)
with _transport_lock:
transport = _transports.get(proxy)
if transport is None:
transport = PyqwestTransport(
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,
# Redirects belong to the httpx client above (which the
# generated clients leave off), not to reqwest.
follow_redirects=False,
),
max_retries=connection_retries,
)
)
transport = PyqwestTransport(retrying_http_transport(proxy))
_transports[proxy] = transport
return transport
class EnvdTransportWithLogger(httpx.HTTPTransport):
_thread_local = threading.local()
@property
def pool(self):
return self._pool
# One transport per (proxy, streaming) pair, separate from the REST API
# pools — envd traffic goes to per-sandbox hosts.
_envd_transports: Dict[Tuple[Optional[ProxyConfig], bool], PyqwestTransport] = {}
def get_envd_transport(
config: ConnectionConfig, http2: bool = True
) -> EnvdTransportWithLogger:
instances: Dict[TransportKey, EnvdTransportWithLogger] = getattr(
EnvdTransportWithLogger._thread_local, "instances", {}
)
key: TransportKey = (http2, config.proxy)
cached = instances.get(key)
if cached is not None:
return cached
config: ConnectionConfig, *, for_streaming: bool = False
) -> PyqwestTransport:
"""The shared pyqwest-backed httpx transports for the envd HTTP API
(file transfers, health checks).
transport = EnvdTransportWithLogger(
limits=limits,
proxy=config.proxy,
http2=http2,
retries=connection_retries,
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 rather than idle bounds. 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 = _envd_transports.get(key)
if transport is None:
transport = PyqwestTransport(
retrying_http_transport(
proxy,
read_timeout=READ_TIMEOUT if for_streaming else None,
)
)
_envd_transports[key] = transport
return transport
def get_envd_api(
config: ConnectionConfig, base_url: str, *, for_streaming: bool = False
) -> httpx.Client:
"""An httpx client for a sandbox's envd HTTP API (file transfers, health
checks) on the shared pyqwest transports. The client itself is a cheap
stateless wrapper — one per consumer is fine — while the pooled transport
underneath is shared and thread-safe."""
return httpx.Client(
base_url=base_url,
transport=get_envd_transport(config, for_streaming=for_streaming),
headers=config.sandbox_headers,
event_hooks=make_logging_event_hooks(config.logger),
)
instances[key] = transport
EnvdTransportWithLogger._thread_local.instances = instances
return transport
packages/python-sdk/e2b/connection_config.py
+7
View File
@@ -21,6 +21,13 @@ narrows it to what the pyqwest REST transports take.
REQUEST_TIMEOUT: float = 60.0 # 60 seconds
# Idle bound for every read on the streaming envd file-transfer transport:
# 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 stream. Matches the previous default stream idle timeout
# (the request timeout).
READ_TIMEOUT: float = 60.0 # 60 seconds
KEEPALIVE_PING_INTERVAL_SEC = 50 # 50 seconds
KEEPALIVE_PING_HEADER = "Keepalive-Ping-Interval"
packages/python-sdk/e2b/envd/client_async/__init__.py
+12 -38
View File
@@ -9,25 +9,20 @@ from typing import (
Callable,
Optional,
TypeVar,
Union,
cast,
)
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from pyqwest import Client, HTTPTransport, Request, Response, Transport
from pyqwest.middleware.retry import RetryTransport
from pyqwest import Client, Request, Response, Transport
from e2b.api import connection_retries
from e2b.api import ProxyConfig, proxy_to_config
from e2b.api.client_async import retrying_http_transport
from e2b.connection_config import ConnectionConfig
from e2b.envd.client_shared import (
ENVD_JSON_CODEC,
ENVD_RPC_COMPRESSION,
plain_http_error,
pool_idle_timeout,
pool_max_idle_per_host,
proxy_to_url,
should_retry_connection,
)
from e2b.envd.interceptors import build_interceptors
from e2b.exceptions import TimeoutException
@@ -37,7 +32,7 @@ TClient = TypeVar("TClient")
_transport_lock = threading.Lock()
# One transport (= one connection pool) per proxy; None is the direct pool.
_transports: dict[Optional[str], "PlainHTTPErrorTransport"] = {}
_transports: dict[Optional[ProxyConfig], "PlainHTTPErrorTransport"] = {}
class PlainHTTPErrorTransport:
@@ -70,37 +65,16 @@ class PlainHTTPErrorTransport:
raise error
class ConnectionRetryTransport(RetryTransport):
"""Retry only failures establishing the connection; see
:func:`e2b.envd.client_shared.should_retry_connection` for the policy
rationale."""
def should_retry_response(
self, request: Request, response: Union[Response, Exception]
) -> bool:
return should_retry_connection(response)
def get_transport(proxy_url: Optional[str]) -> "PlainHTTPErrorTransport":
def get_transport(proxy: Optional[ProxyConfig]) -> "PlainHTTPErrorTransport":
with _transport_lock:
transport = _transports.get(proxy_url)
transport = _transports.get(proxy)
if transport is None:
# connectrpc arms the per-call deadline around the transport, so
# retry backoff counts against the request timeout. The plain-
# error normalization sits outside the retries so it converts
# the settled response once.
transport = PlainHTTPErrorTransport(
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,
),
max_retries=connection_retries,
)
)
_transports[proxy_url] = transport
transport = PlainHTTPErrorTransport(retrying_http_transport(proxy))
_transports[proxy] = transport
return transport
@@ -111,11 +85,11 @@ def create_rpc_client(
) -> TClient:
"""Build a generated async connectrpc client (e.g. ``ProcessClient``)
wired with the shared pyqwest transport (which retries failed connects,
see :class:`ConnectionRetryTransport`), the envd JSON codec, and the
SDK's default-header and logging interceptors. Compression is disabled
(see ``ENVD_RPC_COMPRESSION``).
see :class:`e2b.api.client_async.ConnectionRetryTransport`), the envd
JSON codec, and the SDK's default-header and logging interceptors.
Compression is disabled (see ``ENVD_RPC_COMPRESSION``).
"""
http_client = Client(get_transport(proxy_to_url(config.proxy)))
http_client = Client(get_transport(proxy_to_config(config.proxy)))
return client_cls(
base_url,
codec=ENVD_JSON_CODEC,
packages/python-sdk/e2b/envd/client_shared.py
+6 -69
View File
@@ -1,11 +1,12 @@
"""envd RPC client plumbing shared by the sync and async flavors.
The envd RPC clients (process, filesystem) run on `connectrpc`, whose HTTP
layer is `pyqwest` (Rust reqwest/hyper). This is a separate stack from the
`httpx` transports in `e2b.api`, which keep serving the REST API and the
multipart file transfer endpoints. Unlike the previous httpcore-based
transport, hyper sends RST_STREAM when a server stream is closed early, so
abandoned command/watch streams don't leak on the shared HTTP/2 connection.
layer is `pyqwest` (Rust reqwest/hyper) — built on the same
`retrying_http_transport` pieces as the REST API client in `e2b.api`, in a
separately cached pool. Only the multipart file transfer endpoints stay on
the `httpx` envd transports. Unlike the previous httpcore-based transport,
hyper sends RST_STREAM when a server stream is closed early, so abandoned
command/watch streams don't leak on the shared HTTP/2 connection.
The flavor-specific transports and client factories live in
:mod:`e2b.envd.client_sync` and :mod:`e2b.envd.client_async`, mirroring the
@@ -15,22 +16,12 @@ protobuf codegen (`make generate-envd`).
"""
import json
import os
from typing import Optional, TypedDict, TypeVar
import httpx
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from protobuf import Message
from e2b.exceptions import InvalidArgumentException
# Mirror the httpx pool tuning in `e2b.api.limits` with pyqwest's equivalents.
# `pool_max_idle_per_host` is per host rather than httpx's global idle cap,
# which suits envd traffic — each sandbox is its own host.
pool_idle_timeout = float(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300")
pool_max_idle_per_host = int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20")
_MESSAGE = TypeVar("_MESSAGE", bound=Message)
@@ -133,25 +124,6 @@ def plain_http_error(
)
def should_retry_connection(response: object) -> bool:
"""Whether a transport result is a retryable connection-establishment
failure — the shared policy of the flavor ``ConnectionRetryTransport``s.
pyqwest raises the builtin ``ConnectionError`` only before the request
was written, so retrying exactly these failures can never replay a
request envd may have received — which could re-run a command or
re-deliver events — for unary and streaming RPCs alike. Anything later
(``WriteError``/``ReadError``/``StreamError``, error responses) surfaces
to the caller; the retry middleware's default policy would otherwise also
retry I/O errors and 429/5xx responses for idempotent methods. This
replaces httpcore's transport ``retries`` from the previous stack and
deliberately drops the vendored client's retry on connections dropped
mid-request, which could re-execute a delivered unary RPC like
``SendInput``.
"""
return isinstance(response, ConnectionError)
class _RPCCompression(TypedDict):
send_compression: None
accept_compression: "tuple[()]"
@@ -168,38 +140,3 @@ ENVD_RPC_COMPRESSION: _RPCCompression = {
"send_compression": None,
"accept_compression": (),
}
def proxy_to_url(proxy: object) -> Optional[str]:
"""Narrow the ``proxy`` connection option to the proxy URL string pyqwest
transports take (scheme http, https, socks5, or socks5h, credentials in
the URL userinfo). ``httpx.URL`` and ``httpx.Proxy`` — which the REST
client accepts and the vendored envd client used to — are converted when
they reduce to such a URL; ``httpx.Proxy`` extras that don't (custom
headers, an ssl_context) are rejected rather than silently dropped.
"""
if proxy is None:
return None
if isinstance(proxy, str):
return proxy
if isinstance(proxy, httpx.URL):
return str(proxy)
if isinstance(proxy, httpx.Proxy):
if proxy.headers:
raise InvalidArgumentException(
"Sandbox RPC calls don't support httpx.Proxy custom headers; "
"pass credentials in the proxy URL instead, "
'e.g. proxy="http://user:pass@localhost:8030"'
)
if proxy.ssl_context is not None:
raise InvalidArgumentException(
"Sandbox RPC calls don't support httpx.Proxy ssl_context"
)
url = proxy.url
if proxy.auth is not None:
url = url.copy_with(username=proxy.auth[0], password=proxy.auth[1])
return str(url)
raise InvalidArgumentException(
"Sandbox RPC calls support only URL-string proxies, "
'e.g. proxy="http://user:pass@localhost:8030"'
)
packages/python-sdk/e2b/envd/client_sync/__init__.py
+14 -39
View File
@@ -1,27 +1,22 @@
"""Sync envd RPC clients: shared pyqwest transports and client factory."""
import threading
from typing import Any, Callable, Generator, Iterator, Optional, TypeVar, Union, cast
from typing import Any, Callable, Generator, Iterator, Optional, TypeVar, cast
from pyqwest import (
SyncClient,
SyncHTTPTransport,
SyncRequest,
SyncResponse,
SyncTransport,
)
from pyqwest.middleware.retry import SyncRetryTransport
from e2b.api import connection_retries
from e2b.api import ProxyConfig, proxy_to_config
from e2b.api.client_sync import retrying_http_transport
from e2b.connection_config import ConnectionConfig
from e2b.envd.client_shared import (
ENVD_JSON_CODEC,
ENVD_RPC_COMPRESSION,
plain_http_error,
pool_idle_timeout,
pool_max_idle_per_host,
proxy_to_url,
should_retry_connection,
)
from e2b.envd.interceptors import build_interceptors
@@ -30,7 +25,7 @@ TClient = TypeVar("TClient")
_transport_lock = threading.Lock()
# One transport (= one connection pool) per proxy; None is the direct pool.
_transports: dict[Optional[str], "PlainHTTPErrorTransport"] = {}
_transports: dict[Optional[ProxyConfig], "PlainHTTPErrorTransport"] = {}
class PlainHTTPErrorTransport:
@@ -63,37 +58,16 @@ class PlainHTTPErrorTransport:
raise error
class ConnectionRetryTransport(SyncRetryTransport):
"""Retry only failures establishing the connection; see
:func:`e2b.envd.client_shared.should_retry_connection` for the policy
rationale."""
def should_retry_response(
self, request: SyncRequest, response: Union[SyncResponse, Exception]
) -> bool:
return should_retry_connection(response)
def get_transport(proxy_url: Optional[str]) -> "PlainHTTPErrorTransport":
def get_transport(proxy: Optional[ProxyConfig]) -> "PlainHTTPErrorTransport":
with _transport_lock:
transport = _transports.get(proxy_url)
transport = _transports.get(proxy)
if transport is None:
# connectrpc arms the per-call deadline around the transport, so
# retry backoff counts against the request timeout. The plain-
# error normalization sits outside the retries so it converts
# the settled response once.
transport = PlainHTTPErrorTransport(
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,
),
max_retries=connection_retries,
)
)
_transports[proxy_url] = transport
transport = PlainHTTPErrorTransport(retrying_http_transport(proxy))
_transports[proxy] = transport
return transport
@@ -104,12 +78,13 @@ def create_rpc_client(
) -> TClient:
"""Build a generated sync connectrpc client (e.g. ``ProcessClientSync``)
wired with the shared pyqwest transport (which retries failed connects,
see :class:`ConnectionRetryTransport`), the envd JSON codec, and the
SDK's default-header and logging interceptors. Compression is disabled
(see ``ENVD_RPC_COMPRESSION``). The client is stateless per call and its
transport is process-global, so one instance serves all threads.
see :class:`e2b.api.client_sync.ConnectionRetryTransport`), the envd JSON
codec, and the SDK's default-header and logging interceptors. Compression
is disabled (see ``ENVD_RPC_COMPRESSION``). The client is stateless per
call and its transport is process-global, so one instance serves all
threads.
"""
http_client = SyncClient(get_transport(proxy_to_url(config.proxy)))
http_client = SyncClient(get_transport(proxy_to_config(config.proxy)))
return client_cls(
base_url,
codec=ENVD_JSON_CODEC,
packages/python-sdk/e2b/envd/interceptors.py
+7 -3
View File
@@ -105,9 +105,13 @@ class LoggingInterceptor:
streamed message at DEBUG — mirroring the httpx event hooks used for the
REST API and file transfer requests.
Upstreamed to pyqwest as a logging middleware
(https://github.com/curioswitch/pyqwest/pull/192); this interceptor stays
until that ships in a pyqwest release the SDK can depend on.
pyqwest logs requests itself on the ``pyqwest.access`` and ``pyqwest``
loggers, but those are process-wide loggers that cannot carry the
per-sandbox logger this option gives callers, and they see only the HTTP
exchange — not the streamed messages, nor the Connect error code of a
stream that fails inside a ``200 OK`` response. This interceptor is what
serves the ``logger`` option; the pyqwest loggers sit below it as
transport-level diagnostics.
"""
def __init__(self, logger: logging.Logger, base_url: str):
packages/python-sdk/e2b/sandbox/filesystem/filesystem.py
+36 -3
View File
@@ -1,10 +1,11 @@
import asyncio
import gzip
import re
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from io import IOBase, TextIOBase
from typing import IO, AsyncIterator, Dict, Iterator, Optional, Union, TypedDict
from typing import IO, AsyncIterator, Dict, Iterator, List, Optional, Union, TypedDict
import httpx
@@ -219,9 +220,13 @@ class AsyncFileStreamReader(AsyncIterator[bytes]):
...
"""
def __init__(self, response: httpx.Response):
def __init__(self, response: httpx.Response, idle_timeout: Optional[float] = None):
self._response = response
self._iterator = response.aiter_bytes()
# An explicit per-call idle bound, applied around each read with
# `wait_for` (the transport-wide idle read timeout covers the
# default case; see the flavor `read` implementations).
self._idle_timeout = idle_timeout
self._closed = False
def __aiter__(self) -> AsyncIterator[bytes]:
@@ -229,7 +234,15 @@ class AsyncFileStreamReader(AsyncIterator[bytes]):
async def __anext__(self) -> bytes:
try:
return await self._iterator.__anext__()
read = self._iterator.__anext__()
if self._idle_timeout:
return await asyncio.wait_for(read, self._idle_timeout)
return await read
except asyncio.TimeoutError as e:
# `wait_for`'s expiry; keep the documented httpx exception. The
# transport's own idle read timeout already arrives as one.
await self.aclose()
raise httpx.ReadTimeout(str(e)) from e
except BaseException:
# Covers normal end (StopAsyncIteration) and read errors alike.
await self.aclose()
@@ -254,6 +267,10 @@ def _to_httpx_file(file_path: str, file_data: Union[str, bytes, IO]):
if isinstance(file_data, (str, bytes)):
return ("file", (file_path, file_data))
elif isinstance(file_data, TextIOBase):
# httpx raises TypeError for text-mode objects in a multipart field, so
# reading it here is what lets the multipart path accept one at all —
# at the cost of buffering. The octet-stream path streams it instead,
# which is why a file-like entry defaults to that path.
return ("file", (file_path, file_data.read()))
elif isinstance(file_data, IOBase):
return ("file", (file_path, file_data))
@@ -261,6 +278,22 @@ def _to_httpx_file(file_path: str, file_data: Union[str, bytes, IO]):
raise InvalidArgumentException(f"Unsupported data type for file {file_path}")
def multipart_body_is_streamed(files: List[WriteEntry]) -> bool:
"""Whether the multipart body built from ``files`` streams any entry.
Only binary file-like data is handed to httpx as a stream:
:func:`_to_httpx_file` reads text file-like data into memory first, because
httpx rejects text-mode objects in a multipart field ("Multipart file
uploads must be opened in binary mode"). Those uploads therefore stay
bounded by the request timeout, like ``str``/``bytes`` ones. Compare
``to_upload_body``, which streams text and binary alike on the
octet-stream path (``iter_io_chunks`` encodes the chunks)."""
return any(
isinstance(file["data"], IOBase) and not isinstance(file["data"], TextIOBase)
for file in files
)
def to_upload_body(
data: Union[str, bytes, IO],
use_gzip: bool = False,
packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py
+69 -32
View File
@@ -7,6 +7,7 @@ from connectrpc.code import Code
from connectrpc.errors import ConnectError
from packaging.version import Version
from e2b.api.client_async import get_envd_api
from e2b.connection_config import (
KEEPALIVE_PING_HEADER,
KEEPALIVE_PING_INTERVAL_SEC,
@@ -46,6 +47,7 @@ from e2b.sandbox.filesystem.filesystem import (
WriteEntry,
WriteInfo,
_to_httpx_file,
multipart_body_is_streamed,
map_entry_info,
map_file_type,
metadata_to_headers,
@@ -93,6 +95,11 @@ class Filesystem:
self._envd_version = envd_version
self._connection_config = connection_config
self._envd_api = envd_api
# Streamed downloads default to a sibling client whose transport
# carries the idle read timeout (see `get_envd_transport`).
self._envd_api_streaming = get_envd_api(
connection_config, envd_api_url, for_streaming=True
)
self._rpc = create_rpc_client(
filesystem_connect.FilesystemClient,
@@ -157,24 +164,26 @@ class Filesystem:
"""
Read file content as an `AsyncFileStreamReader` (an `AsyncIterator[bytes]`).
The request timeout bounds only the initial handshake—the returned
iterator is not killed by it while being consumed. A stalled stream is
reclaimed by `stream_idle_timeout` (raising `httpx.ReadTimeout`). The
reader releases its connection once fully consumed; if you don't read it
to the end, use it as an async context manager or call `aclose()` for
deterministic cleanup. There is no garbage-collection safety net—an
abandoned stream holds its connection until the idle timeout fires or
the client is closed.
A `request_timeout` set explicitly for this call is the deadline for
the whole transfer; by default the download is not bounded in total.
A stalled stream is reclaimed by `stream_idle_timeout` (raising
`httpx.ReadTimeout`). The reader releases its connection once fully
consumed; if you don't read it to the end, use it as an async context
manager or call `aclose()` for deterministic cleanup. There is no
garbage-collection safety net—an abandoned stream holds its
connection until the idle timeout fires or the client is closed.
:param path: Path to the file
:param user: Run the operation as this user
:param format: Format of the file content—`stream`
:param request_timeout: Timeout for the request in **seconds**
:param request_timeout: Deadline for the whole transfer in **seconds**
:param gzip: Use gzip compression for the request
:param stream_idle_timeout: Idle timeout in **seconds** for the streamed
body—abort if no chunk arrives within this window. Resets on every
chunk, so it bounds a stalled stream without limiting total transfer
time. Defaults to the request timeout; pass `0` to disable.
body—abort if the response head or the next chunk doesn't arrive
within this window. Resets on every chunk, so it bounds a stalled
stream without limiting total transfer time. Defaults to a
transport-wide idle read timeout (60 seconds); pass `0` to
disable.
:return: File content as an `AsyncFileStreamReader`
"""
@@ -205,35 +214,53 @@ class Filesystem:
if format == "stream":
# Stream the response body instead of buffering it in memory.
request = self._envd_api.build_request(
# Through the pyqwest adapter a per-request timeout is a
# whole-request deadline that would kill long downloads, so it is
# sent only when the caller set `request_timeout` explicitly
# (making it the total-transfer deadline). By default a stalled
# stream is bounded by the streaming transport's idle read
# timeout (see `get_envd_transport`); an explicit
# `stream_idle_timeout` is applied per read with `wait_for` on
# the regular transport instead — so values above the transport
# bound aren't capped by it and `0` disables idle bounding
# entirely.
stream_timeout = ConnectionConfig._get_request_timeout(
None, request_timeout
)
client = (
self._envd_api_streaming
if stream_idle_timeout is None
else self._envd_api
)
request = client.build_request(
"GET",
ENVD_API_FILES_ROUTE,
params=params,
headers=headers,
timeout=timeout,
timeout=stream_timeout,
)
try:
r = await self._envd_api.send(request, stream=True)
if stream_idle_timeout:
r = await asyncio.wait_for(
client.send(request, stream=True), stream_idle_timeout
)
else:
r = await client.send(request, stream=True)
except httpx.RemoteProtocolError as e:
raise await ahandle_envd_api_transport_exception_with_health(
e, self._envd_api
)
except asyncio.TimeoutError as e:
# wait_for's expiry; keep the httpx exception the
# streamed-read contract established.
raise httpx.ReadTimeout(str(e)) from e
err = await _ahandle_filesystem_envd_api_exception(r)
if err:
await r.aclose()
raise err
# The request timeout bounds only the initial handshake; httpx's
# per-chunk `read` timeout becomes the idle-read timeout for the body
# (defaults to the request timeout). The timeout dict is shared by
# reference with the transport and read again when iteration starts.
idle_timeout = (
timeout if stream_idle_timeout is None else stream_idle_timeout
)
request.extensions.get("timeout", {})["read"] = idle_timeout or None
return AsyncFileStreamReader(r)
return AsyncFileStreamReader(r, idle_timeout=stream_idle_timeout)
try:
r = await self._envd_api.get(
@@ -277,7 +304,7 @@ class Filesystem:
:param user: Run the operation as this user
:param request_timeout: Timeout for the request in **seconds**
:param gzip: Use gzip compression for the upload. Implies the `application/octet-stream` upload. Requires envd 0.5.7 or later — when not supported, the upload falls back to uncompressed `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when `data` is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when `data` is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`, which reads text-mode file-like data into memory (httpx only streams binary file objects in a multipart body).
:param metadata: User-defined metadata to persist on the uploaded file as extended attributes. Keys are lowercased by the sandbox; invalid keys or values raise an `InvalidArgumentException`. Requires envd 0.6.2 or later.
:return: Information about the written file
@@ -317,7 +344,7 @@ class Filesystem:
:param user: Run the operation as this user
:param request_timeout: Timeout for the request
:param gzip: Use gzip compression for the upload. Implies the `application/octet-stream` upload. Requires envd 0.5.7 or later — when not supported, the upload falls back to uncompressed `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when any entry is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when any entry is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`, which reads text-mode file-like data into memory (httpx only streams binary file objects in a multipart body).
:param metadata: User-defined metadata to persist on each uploaded file as extended attributes; the same map is applied to every file. Keys are lowercased by the sandbox; invalid keys or values raise an `InvalidArgumentException`. Requires envd 0.6.2 or later.
:return: Information about the written files
"""
@@ -351,9 +378,12 @@ class Filesystem:
# requesting gzip implies it when envd supports it.
use_octet_stream = (use_octet_stream or gzip) and supports_octet_stream
# Each chunk send is bounded by the request timeout (httpx applies it
# per write); a stalled upload the per-write timeout can't observe is
# bounded server-side (envd's per-read idle timeout, envd >= 0.6.7).
# A buffered upload is bounded by the request timeout as a
# whole-request deadline, matching the JS SDK. A streamed (file-like)
# upload carries no client-side timeout — a deadline would kill any
# transfer outlasting it, and a stalled producer is the caller's own
# code — so a stuck streamed upload is bounded server-side (envd's
# per-read idle timeout, envd >= 0.6.7), also matching the JS SDK.
upload_timeout = self._connection_config.get_request_timeout(request_timeout)
# Metadata is sent as request-scoped X-Metadata-* headers, so the same
@@ -375,13 +405,14 @@ class Filesystem:
if gzip:
headers["Content-Encoding"] = "gzip"
is_streamed = not isinstance(file_data, (str, bytes))
try:
r = await self._envd_api.post(
ENVD_API_FILES_ROUTE,
content=to_upload_body_async(file_data, gzip),
headers=headers,
params=params,
timeout=upload_timeout,
timeout=None if is_streamed else upload_timeout,
)
except httpx.RemoteProtocolError as e:
raise await ahandle_envd_api_transport_exception_with_health(
@@ -424,7 +455,13 @@ class Filesystem:
files=httpx_files,
params=params,
headers=extra_headers,
timeout=upload_timeout,
# Only a streamed entry drops the deadline: httpx
# forwards binary `IOBase` entries in chunks, while text
# file-like data was buffered by `_to_httpx_file` (httpx
# rejects text-mode objects in multipart).
timeout=(
None if multipart_body_is_streamed(files) else upload_timeout
),
)
except httpx.RemoteProtocolError as e:
raise await ahandle_envd_api_transport_exception_with_health(
packages/python-sdk/e2b/sandbox_async/main.py
+2 -9
View File
@@ -10,9 +10,8 @@ import httpx
from packaging.version import Version
from typing_extensions import Self, Unpack
from e2b.api import make_async_logging_event_hooks
from e2b.api.client.types import Unset
from e2b.api.client_async import get_envd_transport as get_transport
from e2b.api.client_async import get_envd_api
from e2b.connection_config import ApiParams, ConnectionConfig
from e2b.envd.api import ENVD_API_HEALTH_ROUTE, ahandle_envd_api_exception
from e2b.envd.versions import ENVD_DEBUG_FALLBACK
@@ -108,13 +107,7 @@ class AsyncSandbox(SandboxApi):
"""
super().__init__(**opts)
self._transport = get_transport(self.connection_config)
self._envd_api = httpx.AsyncClient(
base_url=self.envd_api_url,
transport=self._transport,
headers=self.connection_config.sandbox_headers,
event_hooks=make_async_logging_event_hooks(self.connection_config.logger),
)
self._envd_api = get_envd_api(self.connection_config, self.envd_api_url)
self._filesystem = Filesystem(
self.envd_api_url,
self._envd_version,
packages/python-sdk/e2b/sandbox_sync/commands/command.py
+4 -24
View File
@@ -1,12 +1,9 @@
import threading
from typing import Callable, Dict, List, Literal, Optional, Union, overload
import httpx
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from packaging.version import Version
from e2b.api import make_logging_event_hooks
from e2b.api.client_sync import get_envd_transport
from e2b.api.client_sync import get_envd_api
from e2b.connection_config import (
ConnectionConfig,
Username,
@@ -45,31 +42,14 @@ class Commands:
self._envd_api_url = envd_api_url
self._connection_config = connection_config
self._envd_version = envd_version
self._thread_local = threading.local()
self._rpc = create_rpc_client(
process_connect.ProcessClientSync,
envd_api_url,
connection_config,
)
def _create_envd_api(self) -> httpx.Client:
transport = get_envd_transport(self._connection_config)
return httpx.Client(
base_url=self._envd_api_url,
transport=transport,
headers=self._connection_config.sandbox_headers,
event_hooks=make_logging_event_hooks(self._connection_config.logger),
)
@property
def _envd_api(self) -> httpx.Client:
# Unlike the shared RPC client, the httpx transports are per-thread
# (see e2b.api.client_sync), so the client wrapping them is too.
envd_api = getattr(self._thread_local, "envd_api", None)
if envd_api is None:
envd_api = self._create_envd_api()
self._thread_local.envd_api = envd_api
return envd_api
# Like the RPC client, the pyqwest transport underneath is
# thread-safe, so one client serves all threads.
self._envd_api = get_envd_api(connection_config, envd_api_url)
def _check_health(self) -> Optional[bool]:
return check_sandbox_health(self._envd_api)
packages/python-sdk/e2b/sandbox_sync/commands/pty.py
+4 -25
View File
@@ -1,13 +1,9 @@
import httpx
import threading
from typing import Dict, Optional
from connectrpc.code import Code
from connectrpc.errors import ConnectError
from packaging.version import Version
from e2b.api import make_logging_event_hooks
from e2b.api.client_sync import get_envd_transport
from e2b.api.client_sync import get_envd_api
from protobuf import Oneof
from e2b.envd.process import process_connect, process_pb
@@ -43,31 +39,14 @@ class Pty:
self._envd_api_url = envd_api_url
self._connection_config = connection_config
self._envd_version = envd_version
self._thread_local = threading.local()
self._rpc = create_rpc_client(
process_connect.ProcessClientSync,
envd_api_url,
connection_config,
)
def _create_envd_api(self) -> httpx.Client:
transport = get_envd_transport(self._connection_config)
return httpx.Client(
base_url=self._envd_api_url,
transport=transport,
headers=self._connection_config.sandbox_headers,
event_hooks=make_logging_event_hooks(self._connection_config.logger),
)
@property
def _envd_api(self) -> httpx.Client:
# Unlike the shared RPC client, the httpx transports are per-thread
# (see e2b.api.client_sync), so the client wrapping them is too.
envd_api = getattr(self._thread_local, "envd_api", None)
if envd_api is None:
envd_api = self._create_envd_api()
self._thread_local.envd_api = envd_api
return envd_api
# Like the RPC client, the pyqwest transport underneath is
# thread-safe, so one client serves all threads.
self._envd_api = get_envd_api(connection_config, envd_api_url)
def _check_health(self) -> Optional[bool]:
return check_sandbox_health(self._envd_api)
packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py
+49 -52
View File
@@ -1,4 +1,3 @@
import threading
from typing import IO, Dict, List, Literal, Optional, Union, overload
import httpx
@@ -6,8 +5,7 @@ from connectrpc.code import Code
from connectrpc.errors import ConnectError
from packaging.version import Version
from e2b.api import make_logging_event_hooks
from e2b.api.client_sync import get_envd_transport
from e2b.api.client_sync import get_envd_api
from e2b.connection_config import (
KEEPALIVE_PING_HEADER,
KEEPALIVE_PING_INTERVAL_SEC,
@@ -46,6 +44,7 @@ from e2b.sandbox.filesystem.filesystem import (
WriteEntry,
WriteInfo,
_to_httpx_file,
multipart_body_is_streamed,
map_entry_info,
map_file_type,
metadata_to_headers,
@@ -88,32 +87,19 @@ class Filesystem:
self._envd_api_url = envd_api_url
self._envd_version = envd_version
self._connection_config = connection_config
self._thread_local = threading.local()
self._rpc = create_rpc_client(
filesystem_connect.FilesystemClientSync,
envd_api_url,
connection_config,
)
def _create_envd_api(self) -> httpx.Client:
transport = get_envd_transport(self._connection_config)
return httpx.Client(
base_url=self._envd_api_url,
transport=transport,
headers=self._connection_config.sandbox_headers,
event_hooks=make_logging_event_hooks(self._connection_config.logger),
# Like the RPC client, the pyqwest transports underneath are
# thread-safe, so one client (and its streaming sibling, whose
# transport carries the idle read timeout) serves all threads.
self._envd_api = get_envd_api(connection_config, envd_api_url)
self._envd_api_streaming = get_envd_api(
connection_config, envd_api_url, for_streaming=True
)
@property
def _envd_api(self) -> httpx.Client:
# Unlike the shared RPC client, the httpx transports are per-thread
# (see e2b.api.client_sync), so the client wrapping them is too.
envd_api = getattr(self._thread_local, "envd_api", None)
if envd_api is None:
envd_api = self._create_envd_api()
self._thread_local.envd_api = envd_api
return envd_api
@overload
def read(
self,
@@ -171,22 +157,23 @@ class Filesystem:
"""
Read file content as a `FileStreamReader` (an `Iterator[bytes]`).
The request timeout bounds only the initial handshake—the returned
iterator is not killed by it while being consumed. A stalled stream is
reclaimed by `stream_idle_timeout` (raising `httpx.ReadTimeout`). The
reader releases its connection once fully consumed; if you don't read it
to the end, use it as a context manager or call `close()` for
deterministic cleanup.
A `request_timeout` set explicitly for this call is the deadline for
the whole transfer; by default the download is not bounded in total.
A stalled stream is reclaimed by a transport-wide idle read timeout
(raising `httpx.ReadTimeout`). The reader releases its connection
once fully consumed; if you don't read it to the end, use it as a
context manager or call `close()` for deterministic cleanup.
:param path: Path to the file
:param user: Run the operation as this user
:param format: Format of the file content—`stream`
:param request_timeout: Timeout for the request in **seconds**
:param request_timeout: Deadline for the whole transfer in **seconds**
:param gzip: Use gzip compression for the request
:param stream_idle_timeout: Idle timeout in **seconds** for the streamed
body—abort if no chunk arrives within this window. Resets on every
chunk, so it bounds a stalled stream without limiting total transfer
time. Defaults to the request timeout; pass `0` to disable.
:param stream_idle_timeout: Ignored — the sync client cannot
interrupt a blocking read. A stalled streamed read is bounded by
a transport-wide idle read timeout instead (60 seconds), which
resets on every chunk. (`AsyncSandbox.files.read` honors this
parameter.)
:return: File content as a `FileStreamReader`
"""
@@ -217,15 +204,24 @@ class Filesystem:
if format == "stream":
# Stream the response body instead of buffering it in memory.
request = self._envd_api.build_request(
# Through the pyqwest adapter a per-request timeout is a
# whole-request deadline that would kill long downloads, so it is
# sent only when the caller set `request_timeout` explicitly
# (making it the total-transfer deadline). A stalled stream is
# instead bounded by the streaming transport's idle read timeout
# (see `get_envd_transport`), which resets on every chunk.
stream_timeout = ConnectionConfig._get_request_timeout(
None, request_timeout
)
request = self._envd_api_streaming.build_request(
"GET",
ENVD_API_FILES_ROUTE,
params=params,
headers=headers,
timeout=timeout,
timeout=stream_timeout,
)
try:
r = self._envd_api.send(request, stream=True)
r = self._envd_api_streaming.send(request, stream=True)
except httpx.RemoteProtocolError as e:
raise handle_envd_api_transport_exception_with_health(e, self._envd_api)
@@ -234,15 +230,6 @@ class Filesystem:
r.close()
raise err
# The request timeout bounds only the initial handshake; httpx's
# per-chunk `read` timeout becomes the idle-read timeout for the body
# (defaults to the request timeout). The timeout dict is shared by
# reference with the transport and read again when iteration starts.
idle_timeout = (
timeout if stream_idle_timeout is None else stream_idle_timeout
)
request.extensions.get("timeout", {})["read"] = idle_timeout or None
return FileStreamReader(r)
try:
@@ -285,7 +272,7 @@ class Filesystem:
:param user: Run the operation as this user
:param request_timeout: Timeout for the request in **seconds**
:param gzip: Use gzip compression for the upload. Implies the `application/octet-stream` upload. Requires envd 0.5.7 or later — when not supported, the upload falls back to uncompressed `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when `data` is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when `data` is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`, which reads text-mode file-like data into memory (httpx only streams binary file objects in a multipart body).
:param metadata: User-defined metadata to persist on the uploaded file as extended attributes. Keys are lowercased by the sandbox; invalid keys or values raise an `InvalidArgumentException`. Requires envd 0.6.2 or later.
:return: Information about the written file
@@ -325,7 +312,7 @@ class Filesystem:
:param user: Run the operation as this user
:param request_timeout: Timeout for the request
:param gzip: Use gzip compression for the upload. Implies the `application/octet-stream` upload. Requires envd 0.5.7 or later — when not supported, the upload falls back to uncompressed `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when any entry is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`.
:param use_octet_stream: Upload using `application/octet-stream` instead of `multipart/form-data`. Defaults to `None`, which uses octet-stream when any entry is a file-like object (so streamed uploads aren't buffered) and `multipart/form-data` otherwise. Requires envd 0.5.7 or later — when not supported, the upload falls back to `multipart/form-data`, which reads text-mode file-like data into memory (httpx only streams binary file objects in a multipart body).
:param metadata: User-defined metadata to persist on each uploaded file as extended attributes; the same map is applied to every file. Keys are lowercased by the sandbox; invalid keys or values raise an `InvalidArgumentException`. Requires envd 0.6.2 or later.
:return: Information about the written files
"""
@@ -359,9 +346,12 @@ class Filesystem:
# requesting gzip implies it when envd supports it.
use_octet_stream = (use_octet_stream or gzip) and supports_octet_stream
# Each chunk send is bounded by the request timeout (httpx applies it
# per write); a stalled upload the per-write timeout can't observe is
# bounded server-side (envd's per-read idle timeout, envd >= 0.6.7).
# A buffered upload is bounded by the request timeout as a
# whole-request deadline, matching the JS SDK. A streamed (file-like)
# upload carries no client-side timeout — a deadline would kill any
# transfer outlasting it, and a stalled producer is the caller's own
# code — so a stuck streamed upload is bounded server-side (envd's
# per-read idle timeout, envd >= 0.6.7), also matching the JS SDK.
upload_timeout = self._connection_config.get_request_timeout(request_timeout)
# Metadata is sent as request-scoped X-Metadata-* headers, so the same
@@ -382,13 +372,14 @@ class Filesystem:
if gzip:
headers["Content-Encoding"] = "gzip"
is_streamed = not isinstance(file_data, (str, bytes))
try:
r = self._envd_api.post(
ENVD_API_FILES_ROUTE,
content=to_upload_body(file_data, gzip),
headers=headers,
params=params,
timeout=upload_timeout,
timeout=None if is_streamed else upload_timeout,
)
except httpx.RemoteProtocolError as e:
raise handle_envd_api_transport_exception_with_health(
@@ -425,7 +416,13 @@ class Filesystem:
files=httpx_files,
params=params,
headers=extra_headers,
timeout=upload_timeout,
# Only a streamed entry drops the deadline: httpx
# forwards binary `IOBase` entries in chunks, while text
# file-like data was buffered by `_to_httpx_file` (httpx
# rejects text-mode objects in multipart).
timeout=(
None if multipart_body_is_streamed(files) else upload_timeout
),
)
except httpx.RemoteProtocolError as e:
raise handle_envd_api_transport_exception_with_health(e, self._envd_api)
packages/python-sdk/e2b/volume/client_async/__init__.py
+1 -2
View File
@@ -98,8 +98,7 @@ def get_streaming_transport(
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.
whole-request deadlines rather than idle bounds.
"""
return _transport(config, read_timeout=READ_TIMEOUT)
packages/python-sdk/e2b/volume/client_sync/__init__.py
+1 -2
View File
@@ -93,8 +93,7 @@ def get_streaming_transport(config: VolumeConnectionConfig) -> PyqwestTransport:
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.
whole-request deadlines rather than idle bounds.
"""
return _transport(config, read_timeout=READ_TIMEOUT)
packages/python-sdk/e2b/volume/volume_async.py
+3 -5
View File
@@ -529,8 +529,8 @@ class AsyncVolume:
return await asyncio.wait_for(awaitable, stream_idle_timeout)
async def stream_file() -> AsyncIterator[bytes]:
# `read_bounded` expires as a bare timeout; keep the httpx
# exception the streamed-read contract established (the
# `read_bounded` expires as a `wait_for` timeout; keep the
# httpx exception the streamed-read contract established (the
# transport-wide bound already surfaces as one).
try:
stream_cm = stream_client.get_async_httpx_client().stream(
@@ -562,9 +562,7 @@ class AsyncVolume:
yield chunk
finally:
await stream_cm.__aexit__(None, None, None)
# asyncio.TimeoutError is distinct from the builtin until 3.11,
# and `wait_for` raises whichever the running version defines.
except (TimeoutError, asyncio.TimeoutError) as e:
except asyncio.TimeoutError as e:
raise httpx.ReadTimeout(str(e)) from e
return stream_file()
packages/python-sdk/tests/async/sandbox_async/test_config_propagation.py
+1 -7
View File
@@ -1,4 +1,3 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
@@ -16,13 +15,8 @@ BASE_HEADERS = {"X-Test": "base"}
def create_sandbox(monkeypatch, api_key: str) -> AsyncSandbox:
dummy_transport = SimpleNamespace(pool=object())
monkeypatch.setattr(
sandbox_async_main, "get_transport", lambda *_args, **_kwargs: dummy_transport
)
monkeypatch.setattr(
sandbox_async_main.httpx, "AsyncClient", lambda *args, **kwargs: object()
sandbox_async_main, "get_envd_api", lambda *_args, **_kwargs: object()
)
monkeypatch.setattr(
sandbox_async_main, "Filesystem", lambda *args, **kwargs: object()
packages/python-sdk/tests/sync/sandbox_sync/test_sync_client_lifecycle.py
+19 -30
View File
@@ -1,11 +1,9 @@
"""Client lifecycle in the sync sandbox modules: the httpx `envd_api` clients
wrap per-thread transports (see `e2b.api.client_sync`) and are bound per
calling thread, while the connectrpc RPC clients are stateless over a
process-global transport and are built once per module and shared across
threads."""
"""Client lifecycle in the sync sandbox modules: the httpx `envd_api`
clients and the connectrpc RPC clients are both cheap stateless wrappers
over shared, process-global pyqwest transports — built once per module and
shared across threads."""
from concurrent.futures import ThreadPoolExecutor
import threading
from types import SimpleNamespace
from unittest.mock import Mock, sentinel
@@ -27,7 +25,7 @@ def run_in_worker_thread(fn):
return executor.submit(fn).result()
def test_sync_sandbox_envd_api_is_bound_per_calling_thread(monkeypatch, test_api_key):
def test_sync_sandbox_envd_api_delegates_to_filesystem(monkeypatch, test_api_key):
config = ConnectionConfig(api_key=test_api_key)
main_api = Mock(spec=httpx.Client)
filesystem = SimpleNamespace(_envd_api=main_api)
@@ -53,28 +51,18 @@ def test_sync_sandbox_envd_api_is_bound_per_calling_thread(monkeypatch, test_api
assert not hasattr(sandbox, "_transport")
def test_sync_filesystem_envd_api_per_thread_rpc_shared(monkeypatch, test_api_key):
def test_sync_filesystem_clients_are_shared_across_threads(monkeypatch, test_api_key):
config = ConnectionConfig(api_key=test_api_key)
main_thread_id = threading.get_ident()
main_transport = object()
worker_transport = object()
main_api = Mock(spec=httpx.Client)
worker_api = Mock(spec=httpx.Client)
shared_api = Mock(spec=httpx.Client)
streaming_api = Mock(spec=httpx.Client)
shared_rpc = sentinel.filesystem_rpc
monkeypatch.setattr(
filesystem_sync,
"get_envd_transport",
lambda *_args, **_kwargs: main_transport
if threading.get_ident() == main_thread_id
else worker_transport,
)
monkeypatch.setattr(
filesystem_sync.httpx,
"Client",
lambda *args, **kwargs: main_api
if kwargs["transport"] is main_transport
else worker_api,
"get_envd_api",
lambda *_args, **kwargs: streaming_api
if kwargs.get("for_streaming")
else shared_api,
)
monkeypatch.setattr(
filesystem_sync,
@@ -88,12 +76,13 @@ def test_sync_filesystem_envd_api_per_thread_rpc_shared(monkeypatch, test_api_ke
config,
)
assert fs._envd_api is main_api
assert fs._envd_api is shared_api
assert fs._envd_api_streaming is streaming_api
assert fs._rpc is shared_rpc
worker_api_result, worker_rpc_result = run_in_worker_thread(
lambda: (fs._envd_api, fs._rpc)
worker_api, worker_streaming, worker_rpc = run_in_worker_thread(
lambda: (fs._envd_api, fs._envd_api_streaming, fs._rpc)
)
assert worker_api_result is worker_api
assert worker_rpc_result is shared_rpc
assert fs._envd_api is main_api
assert worker_api is shared_api
assert worker_streaming is streaming_api
assert worker_rpc is shared_rpc
packages/python-sdk/tests/test_api_client_transport.py
+93 -123
View File
@@ -2,49 +2,36 @@ import asyncio
import base64
import json
import logging
import ssl
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import cast
import httpx
import pytest
from pyqwest import (
Proxy,
Request,
SyncRequest,
SyncTransport,
Transport,
)
from pyqwest.httpx import AsyncPyqwestTransport, PyqwestTransport
import e2b.api.client_async as client_async
import e2b.api.client_sync as client_sync
from e2b.api import ProxyConfig, proxy_to_config
from e2b.api.client_async import AsyncEnvdTransportWithLogger
from e2b.api.client_async import get_api_client as get_async_api_client
from e2b.api.client_async import get_envd_api as get_async_envd_api
from e2b.api.client_async import get_envd_transport as get_async_envd_transport
from e2b.api.client_async import get_transport as get_async_transport
from e2b.api.client_sync import EnvdTransportWithLogger
from e2b.api.client_sync import get_api_client as get_sync_api_client
from e2b.api.client_sync import get_envd_api as get_sync_envd_api
from e2b.api.client_sync import get_envd_transport as get_sync_envd_transport
from e2b.api.client_sync import get_transport as get_sync_transport
from e2b.connection_config import ConnectionConfig, ProxyTypes
from e2b.exceptions import InvalidArgumentException
from e2b.connection_config import ConnectionConfig
def reset_sync_api_transports():
client_sync._transports.clear()
client_sync._envd_transports.clear()
def reset_async_api_transports():
client_async._transports.clear()
def reset_sync_envd_transports():
EnvdTransportWithLogger._thread_local.instances = {}
client_async._envd_transports.clear()
def run_in_worker_thread(fn):
@@ -52,76 +39,6 @@ def run_in_worker_thread(fn):
return executor.submit(fn).result()
def test_proxy_to_config_narrows_urls():
assert proxy_to_config(None) is None
assert proxy_to_config("http://127.0.0.1:9999") == ProxyConfig(
"http://127.0.0.1:9999"
)
assert proxy_to_config(httpx.URL("http://127.0.0.1:9999")) == ProxyConfig(
"http://127.0.0.1:9999"
)
# Guards callers that ignore the type hint.
with pytest.raises(InvalidArgumentException, match="URL-string"):
proxy_to_config(cast(ProxyTypes, object()))
def test_proxy_to_config_converts_httpx_proxy():
# Everything pyqwest's Proxy can express carries over: the URL, the
# credentials httpx.Proxy splits off the URL, and headers for the proxy.
assert proxy_to_config(httpx.Proxy("http://127.0.0.1:9999")) == ProxyConfig(
"http://127.0.0.1:9999"
)
assert proxy_to_config(
httpx.Proxy("http://user:pass@127.0.0.1:9999")
) == ProxyConfig("http://127.0.0.1:9999", auth=("user", "pass"))
assert proxy_to_config(
httpx.Proxy("http://127.0.0.1:9999", auth=("user@x", "p@ss"))
) == ProxyConfig("http://127.0.0.1:9999", auth=("user@x", "p@ss"))
assert proxy_to_config(
httpx.Proxy("http://127.0.0.1:9999", headers={"X-Auth": "t"})
) == ProxyConfig("http://127.0.0.1:9999", headers=(("x-auth", "t"),))
# A per-proxy TLS context has no pyqwest counterpart; rejected rather than
# silently dropped.
with pytest.raises(InvalidArgumentException, match="ssl_context"):
proxy_to_config(
httpx.Proxy(
"https://127.0.0.1:9999", ssl_context=ssl.create_default_context()
)
)
def test_proxy_config_builds_a_pyqwest_proxy():
config = ProxyConfig(
"http://127.0.0.1:9999", auth=("user", "pass"), headers=(("x-auth", "t"),)
)
assert isinstance(config.to_pyqwest(), Proxy)
# Equal configs are one cache key, even though the Proxy objects they
# build compare by identity.
assert config == ProxyConfig(
"http://127.0.0.1:9999", auth=("user", "pass"), headers=(("x-auth", "t"),)
)
assert config.to_pyqwest() != config.to_pyqwest()
def test_connection_retry_policy_retries_only_connection_errors():
# The inner transport is never invoked by should_retry_response.
sync_policy = client_sync.ConnectionRetryTransport(cast(SyncTransport, object()))
async_policy = client_async.ConnectionRetryTransport(cast(Transport, object()))
sync_request = SyncRequest("GET", "https://example.com")
async_request = Request("GET", "https://example.com")
assert sync_policy.should_retry_response(sync_request, ConnectionError("refused"))
# TimeoutError is an OSError but not a ConnectionError: the request
# may have been written, so it must not be replayed.
assert not sync_policy.should_retry_response(sync_request, TimeoutError())
assert not sync_policy.should_retry_response(sync_request, RuntimeError())
assert async_policy.should_retry_response(async_request, ConnectionError("refused"))
assert not async_policy.should_retry_response(async_request, TimeoutError())
assert not async_policy.should_retry_response(async_request, RuntimeError())
def test_sync_api_client_proxy_uses_explicit_transport(test_api_key):
reset_sync_api_transports()
config = ConnectionConfig(
@@ -197,26 +114,48 @@ def test_sync_api_client_request_timeout_zero_disables_timeout(test_api_key):
reset_sync_api_transports()
def test_sync_envd_transport_uses_separate_stack(test_api_key):
# envd file-transfer traffic stays on the httpx-native transport (its RPC
# migration is a separate change); only REST API calls go through pyqwest.
def test_sync_envd_transports_keyed_by_streaming(test_api_key):
# The envd HTTP API pools are separate from the REST API pools, and the
# streaming variant (which carries the idle read timeout) is its own
# pool per proxy.
reset_sync_api_transports()
reset_sync_envd_transports()
config = ConnectionConfig(api_key=test_api_key)
try:
api_transport = get_sync_transport(config)
envd_transport = get_sync_envd_transport(config)
streaming_transport = get_sync_envd_transport(config, for_streaming=True)
assert isinstance(api_transport, PyqwestTransport)
assert isinstance(envd_transport, httpx.HTTPTransport)
assert get_sync_transport(config) is api_transport
assert isinstance(envd_transport, PyqwestTransport)
assert envd_transport is not api_transport
assert streaming_transport is not envd_transport
assert get_sync_envd_transport(config) is envd_transport
envd_pool = envd_transport._pool
assert envd_pool._http2 is True # ty: ignore[possibly-missing-attribute]
assert (
get_sync_envd_transport(config, for_streaming=True) is streaming_transport
)
finally:
reset_sync_api_transports()
reset_sync_envd_transports()
def test_sync_envd_api_client_wiring(test_api_key):
reset_sync_api_transports()
config = ConnectionConfig(api_key=test_api_key, access_token="tok")
client = get_sync_envd_api(config, "https://sandbox.e2b.app")
streaming = get_sync_envd_api(config, "https://sandbox.e2b.app", for_streaming=True)
try:
assert client.base_url == "https://sandbox.e2b.app"
assert client._transport is get_sync_envd_transport(config)
assert streaming._transport is get_sync_envd_transport(
config, for_streaming=True
)
for header, value in config.sandbox_headers.items():
assert client.headers[header] == value
finally:
client.close()
streaming.close()
reset_sync_api_transports()
def test_sync_api_client_is_shared_across_threads(test_api_key):
@@ -238,23 +177,6 @@ def test_sync_api_client_is_shared_across_threads(test_api_key):
reset_sync_api_transports()
def test_sync_envd_transport_cache_is_thread_local(test_api_key):
reset_sync_envd_transports()
config = ConnectionConfig(api_key=test_api_key)
try:
main_transport = get_sync_envd_transport(config)
thread_transport = run_in_worker_thread(lambda: get_sync_envd_transport(config))
assert main_transport is get_sync_envd_transport(config)
assert thread_transport is not main_transport
main_pool = main_transport._pool
assert main_pool._http2 is True # ty: ignore[possibly-missing-attribute]
assert thread_transport._pool._http2 is True
finally:
reset_sync_envd_transports()
@pytest.mark.asyncio
async def test_async_api_client_proxy_uses_explicit_transport(test_api_key):
reset_async_api_transports()
@@ -325,24 +247,41 @@ async def test_async_api_client_is_shared_across_loops(test_api_key):
@pytest.mark.asyncio
async def test_async_envd_transport_uses_separate_stack(test_api_key):
async def test_async_envd_transports_keyed_by_streaming(test_api_key):
reset_async_api_transports()
AsyncEnvdTransportWithLogger._instances.clear()
config = ConnectionConfig(api_key=test_api_key)
try:
api_transport = get_async_transport(config)
envd_transport = get_async_envd_transport(config)
streaming_transport = get_async_envd_transport(config, for_streaming=True)
assert isinstance(api_transport, AsyncPyqwestTransport)
assert isinstance(envd_transport, httpx.AsyncHTTPTransport)
assert get_async_transport(config) is api_transport
assert isinstance(envd_transport, AsyncPyqwestTransport)
assert envd_transport is not api_transport
assert streaming_transport is not envd_transport
assert get_async_envd_transport(config) is envd_transport
envd_pool = envd_transport._pool
assert envd_pool._http2 is True # ty: ignore[possibly-missing-attribute]
assert (
get_async_envd_transport(config, for_streaming=True) is streaming_transport
)
finally:
reset_async_api_transports()
AsyncEnvdTransportWithLogger._instances.clear()
@pytest.mark.asyncio
async def test_async_envd_api_client_wiring(test_api_key):
reset_async_api_transports()
config = ConnectionConfig(api_key=test_api_key, access_token="tok")
client = get_async_envd_api(config, "https://sandbox.e2b.app")
try:
assert client.base_url == "https://sandbox.e2b.app"
assert client._transport is get_async_envd_transport(config)
for header, value in config.sandbox_headers.items():
assert client.headers[header] == value
finally:
await client.aclose()
reset_async_api_transports()
class _EchoHandler(BaseHTTPRequestHandler):
@@ -372,6 +311,16 @@ class _EchoHandler(BaseHTTPRequestHandler):
return
self.wfile.write(body)
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
received = self.rfile.read(length) if length else b""
body = json.dumps({"received": len(received)}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
@@ -640,3 +589,24 @@ async def test_async_api_client_body_timeout_raises_httpx_read_timeout(
finally:
await httpx_client.aclose()
reset_async_api_transports()
def test_sync_transport_sends_multipart_bodies(test_api_key, echo_server):
# `files=` uploads (envd `files.write`) go out as httpx's MultipartStream,
# which implements both SyncByteStream and AsyncByteStream. The adapter's
# sync path used to match AsyncByteStream first and raise from inside the
# body iterator, surfacing as a WriteError mid-request; pyqwest 0.8 matches
# the sync case first, so the SDK no longer rewraps the stream.
reset_sync_api_transports()
config = ConnectionConfig(api_key=test_api_key)
client = httpx.Client(
base_url=echo_server, transport=client_sync.get_envd_transport(config)
)
try:
response = client.post("/files", files=[("file", ("a.txt", b"x" * 4096))])
assert response.status_code == 200
assert response.json()["received"] > 4096
finally:
client.close()
reset_sync_api_transports()
packages/python-sdk/tests/test_env_var_parsing.py
+7 -15
View File
@@ -1,38 +1,30 @@
"""Numeric E2B_* env vars are parsed at import time; an empty-string value
(e.g. `E2B_MAX_CONNECTIONS=` in a dotenv file) must fall back to the default
(e.g. `E2B_KEEPALIVE_EXPIRY=` in a dotenv file) must fall back to the default
instead of raising ValueError when the module is imported."""
import importlib
import e2b.api
import e2b.envd.client_shared
_ENV_VARS = (
"E2B_KEEPALIVE_EXPIRY",
"E2B_MAX_KEEPALIVE_CONNECTIONS",
"E2B_MAX_CONNECTIONS",
"E2B_CONNECTION_RETRIES",
)
def _reload():
return (
importlib.reload(e2b.envd.client_shared),
importlib.reload(e2b.api),
)
return importlib.reload(e2b.api)
def test_empty_env_vars_fall_back_to_defaults(monkeypatch):
for var in _ENV_VARS:
monkeypatch.setenv(var, "")
try:
client_shared, api = _reload()
assert client_shared.pool_idle_timeout == 300
assert client_shared.pool_max_idle_per_host == 20
api = _reload()
assert api.pool_idle_timeout == 300
assert api.pool_max_idle_per_host == 20
assert api.connection_retries == 3
assert api.limits.max_keepalive_connections == 20
assert api.limits.max_connections == 2000
assert api.limits.keepalive_expiry == 300
finally:
monkeypatch.undo()
_reload()
@@ -42,8 +34,8 @@ def test_set_env_vars_are_honored(monkeypatch):
monkeypatch.setenv("E2B_KEEPALIVE_EXPIRY", "42")
monkeypatch.setenv("E2B_CONNECTION_RETRIES", "5")
try:
client_shared, api = _reload()
assert client_shared.pool_idle_timeout == 42
api = _reload()
assert api.pool_idle_timeout == 42
assert api.connection_retries == 5
finally:
monkeypatch.undo()
packages/python-sdk/tests/test_envd_client_transport.py
+75 -28
View File
@@ -1,8 +1,14 @@
from typing import cast
import httpx
import pytest
from pyqwest import Proxy
import e2b.api.client_async as api_client_async
import e2b.api.client_sync as api_client_sync
from e2b.api import ProxyConfig, proxy_to_config
from e2b.connection_config import ProxyTypes
from e2b.envd import client_async, client_sync
from e2b.envd.client_shared import proxy_to_url
from e2b.exceptions import InvalidArgumentException
@@ -15,73 +21,114 @@ def reset_transport_caches():
client_async._transports.clear()
def test_proxy_to_url_none():
assert proxy_to_url(None) is None
def test_proxy_to_config_none():
assert proxy_to_config(None) is None
def test_proxy_to_url_str():
assert proxy_to_url("http://127.0.0.1:8080") == "http://127.0.0.1:8080"
def test_proxy_to_config_str():
assert proxy_to_config("http://127.0.0.1:8080") == ProxyConfig(
"http://127.0.0.1:8080"
)
def test_proxy_to_url_keeps_credentials_from_url():
assert proxy_to_url("http://user:pass@localhost:8030") == (
def test_proxy_to_config_keeps_credentials_from_url():
assert proxy_to_config("http://user:pass@localhost:8030") == ProxyConfig(
"http://user:pass@localhost:8030"
)
def test_proxy_to_url_converts_httpx_url():
assert proxy_to_url(httpx.URL("http://localhost:8030")) == "http://localhost:8030"
def test_proxy_to_config_converts_httpx_url():
assert proxy_to_config(httpx.URL("http://localhost:8030")) == ProxyConfig(
"http://localhost:8030"
)
def test_proxy_to_url_converts_httpx_proxy_with_auth():
def test_proxy_to_config_converts_httpx_proxy_with_auth():
# Credentials stay a separate (username, password) pair instead of being
# percent-encoded back into the URL userinfo.
proxy = httpx.Proxy("http://localhost:8030", auth=("user@x", "p@ss"))
assert proxy_to_url(proxy) == "http://user%40x:p%40ss@localhost:8030"
assert proxy_to_config(proxy) == ProxyConfig(
"http://localhost:8030", auth=("user@x", "p@ss")
)
def test_proxy_to_url_keeps_credentials_from_httpx_proxy_url():
# httpx.Proxy pulls userinfo out of the URL into `.auth`; conversion
# has to fold it back in.
def test_proxy_to_config_keeps_credentials_from_httpx_proxy_url():
# httpx.Proxy pulls userinfo out of the URL into `.auth`, which is how
# pyqwest takes it too.
proxy = httpx.Proxy("http://user:pass@localhost:8030")
assert proxy_to_url(proxy) == "http://user:pass@localhost:8030"
assert proxy_to_config(proxy) == ProxyConfig(
"http://localhost:8030", auth=("user", "pass")
)
def test_proxy_to_url_rejects_httpx_proxy_headers():
# Typed so `except SandboxException` handlers catch it at the RPC call.
def test_proxy_to_config_carries_httpx_proxy_headers():
proxy = httpx.Proxy("http://localhost:8030", headers={"X-Custom": "1"})
with pytest.raises(InvalidArgumentException, match="headers"):
proxy_to_url(proxy)
assert proxy_to_config(proxy) == ProxyConfig(
"http://localhost:8030", headers=(("x-custom", "1"),)
)
def test_proxy_to_url_rejects_httpx_proxy_ssl_context():
def test_proxy_config_builds_a_pyqwest_proxy():
config = ProxyConfig(
"http://localhost:8030", auth=("user", "pass"), headers=(("x-custom", "1"),)
)
assert isinstance(config.to_pyqwest(), Proxy)
# Equal configs are one cache key even though the Proxy objects they build
# are not equal to each other.
assert config == ProxyConfig(
"http://localhost:8030", auth=("user", "pass"), headers=(("x-custom", "1"),)
)
assert config.to_pyqwest() != config.to_pyqwest()
def test_proxy_to_config_rejects_httpx_proxy_ssl_context():
import ssl
# Typed so `except SandboxException` handlers catch it at the RPC call.
proxy = httpx.Proxy(
"https://localhost:8030", ssl_context=ssl.create_default_context()
)
with pytest.raises(InvalidArgumentException, match="ssl_context"):
proxy_to_url(proxy)
proxy_to_config(proxy)
def test_proxy_to_url_rejects_unknown_types():
def test_proxy_to_config_rejects_unknown_types():
with pytest.raises(InvalidArgumentException, match="URL-string"):
proxy_to_url(object())
proxy_to_config(cast(ProxyTypes, object()))
def test_sync_transport_is_cached_per_proxy():
proxy = ProxyConfig("http://127.0.0.1:8080")
transport_a = client_sync.get_transport(None)
transport_b = client_sync.get_transport(None)
transport_c = client_sync.get_transport("http://127.0.0.1:8080")
transport_d = client_sync.get_transport("http://127.0.0.1:8080")
transport_c = client_sync.get_transport(proxy)
# A second, equal config keys the same pool.
transport_d = client_sync.get_transport(ProxyConfig("http://127.0.0.1:8080"))
assert transport_a is transport_b
assert transport_c is transport_d
assert transport_a is not transport_c
def test_sync_transport_is_not_shared_across_proxy_credentials():
# Same proxy URL, different credentials or headers: separate pools, since
# the proxy configuration is fixed per transport.
url = "http://127.0.0.1:8080"
plain = client_sync.get_transport(ProxyConfig(url))
with_auth = client_sync.get_transport(ProxyConfig(url, auth=("user", "pass")))
with_headers = client_sync.get_transport(
ProxyConfig(url, headers=(("x-custom", "1"),))
)
assert plain is not with_auth
assert plain is not with_headers
assert with_auth is not with_headers
def test_async_transport_is_cached_per_proxy():
transport_a = client_async.get_transport(None)
transport_b = client_async.get_transport(None)
transport_c = client_async.get_transport("http://127.0.0.1:8080")
transport_c = client_async.get_transport(ProxyConfig("http://127.0.0.1:8080"))
assert transport_a is transport_b
assert transport_a is not transport_c
@@ -98,7 +145,7 @@ def test_transport_stack_normalizes_plain_errors_and_retries_connects():
async_transport = client_async.get_transport(None)
assert isinstance(sync_transport, client_sync.PlainHTTPErrorTransport)
assert isinstance(async_transport, client_async.PlainHTTPErrorTransport)
assert isinstance(sync_transport._inner, client_sync.ConnectionRetryTransport)
assert isinstance(async_transport._inner, client_async.ConnectionRetryTransport)
assert isinstance(sync_transport._inner, api_client_sync.ConnectionRetryTransport)
assert isinstance(async_transport._inner, api_client_async.ConnectionRetryTransport)
assert sync_transport._inner._max_retries == connection_retries
assert async_transport._inner._max_retries == connection_retries
packages/python-sdk/tests/test_envd_plain_http_errors.py
+5 -5
View File
@@ -31,12 +31,12 @@ from pyqwest import (
)
import e2b.sandbox_async.commands.command as command_async
from e2b.envd.client_async import (
ConnectionRetryTransport,
PlainHTTPErrorTransport,
)
from e2b.envd.client_sync import (
from e2b.api.client_async import ConnectionRetryTransport
from e2b.api.client_sync import (
ConnectionRetryTransport as SyncConnectionRetryTransport,
)
from e2b.envd.client_async import PlainHTTPErrorTransport
from e2b.envd.client_sync import (
PlainHTTPErrorTransport as SyncPlainHTTPErrorTransport,
)
from e2b.envd.process.process_pb import ConnectRequest
packages/python-sdk/tests/test_envd_retry_transport.py
+2 -2
View File
@@ -24,8 +24,8 @@ from envd_frame_server import (
make_sync_client,
)
from e2b.envd.client_async import ConnectionRetryTransport
from e2b.envd.client_sync import (
from e2b.api.client_async import ConnectionRetryTransport
from e2b.api.client_sync import (
ConnectionRetryTransport as SyncConnectionRetryTransport,
)
from e2b.envd.process.process_pb import ConnectRequest
packages/python-sdk/tests/test_file_stream_reader.py
+27
View File
@@ -222,5 +222,32 @@ async def test_async_abandoned_reader_is_reclaimed_on_client_close():
assert _active_connections(client) == 0
async def test_async_reader_explicit_idle_timeout_bounds_each_read():
# The per-call idle bound is enforced with wait_for around each read, so
# it works on the regular transport (no transport-level read timeout).
async with httpx.AsyncClient() as client:
port = _start_chunked_server(stall_before=1, stall_seconds=0.5)
request = client.build_request("GET", f"http://127.0.0.1:{port}/files")
reader = AsyncFileStreamReader(
await client.send(request, stream=True), idle_timeout=0.05
)
assert await reader.__anext__() == CHUNKS[0]
with pytest.raises(httpx.ReadTimeout):
await reader.__anext__()
assert _active_connections(client) == 0
async def test_async_reader_explicit_idle_timeout_allows_prompt_chunks():
async with httpx.AsyncClient() as client:
port = _start_chunked_server()
request = client.build_request("GET", f"http://127.0.0.1:{port}/files")
reader = AsyncFileStreamReader(
await client.send(request, stream=True), idle_timeout=5.0
)
collected = b"".join([chunk async for chunk in reader])
assert collected == EXPECTED
assert _active_connections(client) == 0
if __name__ == "__main__":
raise SystemExit(pytest.main([__file__, "-v"]))
packages/python-sdk/tests/test_filesystem_models.py
+21
View File
@@ -1,9 +1,11 @@
import datetime
import io
from protobuf.wkt import Timestamp
from e2b.envd.filesystem import filesystem_pb
from e2b.sandbox.filesystem.filesystem import (
multipart_body_is_streamed,
FileType,
WriteInfo,
map_entry_info,
@@ -128,3 +130,22 @@ def test_convert_volume_entry_stat_normalizes_naive_times_to_utc():
assert stat.atime.tzinfo == datetime.timezone.utc
assert stat.mtime.tzinfo == datetime.timezone.utc
assert stat.ctime.tzinfo == datetime.timezone.utc
def test_multipart_body_is_streamed_only_for_binary_file_like_entries():
# The multipart upload drops its request deadline only when httpx really
# streams the body; `_to_httpx_file` reads text file-like data into memory,
# so those uploads must stay bounded like str/bytes ones.
assert not multipart_body_is_streamed([{"path": "a.txt", "data": "text"}])
assert not multipart_body_is_streamed([{"path": "a.bin", "data": b"bytes"}])
assert not multipart_body_is_streamed(
[{"path": "a.txt", "data": io.StringIO("text")}]
)
assert multipart_body_is_streamed([{"path": "a.bin", "data": io.BytesIO(b"bytes")}])
# Any streamed entry makes the whole body streamed.
assert multipart_body_is_streamed(
[
{"path": "a.txt", "data": "text"},
{"path": "b.bin", "data": io.BytesIO(b"bytes")},
]
)