diff --git a/.changeset/python-pyqwest-envd-api.md b/.changeset/python-pyqwest-envd-api.md new file mode 100644 index 0000000000..479e8c4154 --- /dev/null +++ b/.changeset/python-pyqwest-envd-api.md @@ -0,0 +1,27 @@ +--- +"@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. diff --git a/packages/python-sdk/e2b/api/__init__.py b/packages/python-sdk/e2b/api/__init__.py index a6e10900bc..5ab5aeeb89 100644 --- a/packages/python-sdk/e2b/api/__init__.py +++ b/packages/python-sdk/e2b/api/__init__.py @@ -69,11 +69,11 @@ async def on_response(response: Response) -> None: 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). +# Mirror the httpx pool tuning above with pyqwest's equivalents, shared by +# the REST API and envd RPC transports. `pool_max_idle_per_host` is per host +# rather than httpx's global idle cap, which suits both: API traffic goes to +# a single host and each sandbox is its own host. reqwest has no counterpart +# to `E2B_MAX_CONNECTIONS` (it does not cap concurrent connections). 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 +109,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 +118,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"' ) diff --git a/packages/python-sdk/e2b/api/client_async/__init__.py b/packages/python-sdk/e2b/api/client_async/__init__.py index d4de1b49a8..2a11bc17a2 100644 --- a/packages/python-sdk/e2b/api/client_async/__init__.py +++ b/packages/python-sdk/e2b/api/client_async/__init__.py @@ -1,6 +1,5 @@ import asyncio import threading -import weakref from typing import Dict, Optional, Tuple, Union import httpx @@ -13,14 +12,12 @@ 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: @@ -53,12 +50,14 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: 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] @@ -66,74 +65,103 @@ def should_retry_response( 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, + ), + 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], "AsyncApiPyqwestTransport"] = {} def get_transport(config: ConnectionConfig) -> "AsyncApiPyqwestTransport": """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 = AsyncApiPyqwestTransport( - ConnectionRetryTransport( - HTTPTransport( - tls_include_system_certs=True, - proxy=proxy.to_pyqwest() if proxy is not None else None, - pool_idle_timeout=pool_idle_timeout, - pool_max_idle_per_host=pool_max_idle_per_host, - ), - max_retries=connection_retries, - ) - ) + transport = AsyncApiPyqwestTransport(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], "AsyncApiPyqwestTransport" +] = {} 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 - - 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 - - return transport + config: ConnectionConfig, *, for_streaming: bool = False +) -> "AsyncApiPyqwestTransport": + """The shared pyqwest-backed httpx transports for the envd HTTP API + (file transfers, health checks). + + 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 = AsyncApiPyqwestTransport( + 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.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), + ) diff --git a/packages/python-sdk/e2b/api/client_sync/__init__.py b/packages/python-sdk/e2b/api/client_sync/__init__.py index b5ffa06ee3..c31bab65a9 100644 --- a/packages/python-sdk/e2b/api/client_sync/__init__.py +++ b/packages/python-sdk/e2b/api/client_sync/__init__.py @@ -11,14 +11,12 @@ 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: @@ -49,12 +47,14 @@ def handle_request(self, request: httpx.Request) -> httpx.Response: 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] @@ -62,66 +62,101 @@ def should_retry_response( 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, + ), + 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], "ApiPyqwestTransport"] = {} def get_transport(config: ConnectionConfig) -> "ApiPyqwestTransport": """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 = ApiPyqwestTransport( - ConnectionRetryTransport( - SyncHTTPTransport( - tls_include_system_certs=True, - proxy=proxy.to_pyqwest() if proxy is not None else None, - pool_idle_timeout=pool_idle_timeout, - pool_max_idle_per_host=pool_max_idle_per_host, - ), - max_retries=connection_retries, - ) - ) + transport = ApiPyqwestTransport(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], "ApiPyqwestTransport"] = {} 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 - - transport = EnvdTransportWithLogger( - limits=limits, - proxy=config.proxy, - http2=http2, - retries=connection_retries, + config: ConnectionConfig, *, for_streaming: bool = False +) -> "ApiPyqwestTransport": + """The shared pyqwest-backed httpx transports for the envd HTTP API + (file transfers, health checks). + + The streaming transport carries ``read_timeout``, the idle bound on + every read: it resets after each successful read, so it caps how long a + streamed download may stall without limiting total transfer time. It is + fixed per transport — the adapter's per-request timeouts are + whole-request deadlines, and the sync adapter does not bound body reads + at all. Only streamed downloads use it: reqwest's read timer keeps + running while a request body is sent and while waiting for the response + head, so on the regular transport it would cut off uploads and slow + unary responses longer than the idle bound (those stay bounded by their + whole-request deadlines instead). + """ + proxy = proxy_to_config(config.proxy) + key = (proxy, for_streaming) + with _transport_lock: + transport = _envd_transports.get(key) + if transport is None: + transport = ApiPyqwestTransport( + 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 diff --git a/packages/python-sdk/e2b/connection_config.py b/packages/python-sdk/e2b/connection_config.py index 0de3688470..e9f2004d08 100644 --- a/packages/python-sdk/e2b/connection_config.py +++ b/packages/python-sdk/e2b/connection_config.py @@ -21,6 +21,13 @@ 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" diff --git a/packages/python-sdk/e2b/envd/client_async/__init__.py b/packages/python-sdk/e2b/envd/client_async/__init__.py index b1597ea200..36a19bc1ac 100644 --- a/packages/python-sdk/e2b/envd/client_async/__init__.py +++ b/packages/python-sdk/e2b/envd/client_async/__init__.py @@ -9,25 +9,20 @@ 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 @@ _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 @@ async def execute(self, request: Request) -> Response: 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, diff --git a/packages/python-sdk/e2b/envd/client_shared.py b/packages/python-sdk/e2b/envd/client_shared.py index 950f6bb7c2..aa152cd1d6 100644 --- a/packages/python-sdk/e2b/envd/client_shared.py +++ b/packages/python-sdk/e2b/envd/client_shared.py @@ -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,23 +16,12 @@ """ 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.connection_config import ProxyTypes -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) @@ -134,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[()]" @@ -169,38 +140,3 @@ class _RPCCompression(TypedDict): "send_compression": None, "accept_compression": (), } - - -def proxy_to_url(proxy: Optional[ProxyTypes]) -> 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"' - ) diff --git a/packages/python-sdk/e2b/envd/client_sync/__init__.py b/packages/python-sdk/e2b/envd/client_sync/__init__.py index 7ceaf980eb..fc10cf9881 100644 --- a/packages/python-sdk/e2b/envd/client_sync/__init__.py +++ b/packages/python-sdk/e2b/envd/client_sync/__init__.py @@ -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 @@ _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 @@ def execute_sync(self, request: SyncRequest) -> SyncResponse: 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, diff --git a/packages/python-sdk/e2b/envd/interceptors.py b/packages/python-sdk/e2b/envd/interceptors.py index 396bf650c4..b50300afaa 100644 --- a/packages/python-sdk/e2b/envd/interceptors.py +++ b/packages/python-sdk/e2b/envd/interceptors.py @@ -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): diff --git a/packages/python-sdk/e2b/sandbox/filesystem/filesystem.py b/packages/python-sdk/e2b/sandbox/filesystem/filesystem.py index d80c296241..5d77b8e5f3 100644 --- a/packages/python-sdk/e2b/sandbox/filesystem/filesystem.py +++ b/packages/python-sdk/e2b/sandbox/filesystem/filesystem.py @@ -1,3 +1,4 @@ +import asyncio import gzip import re from dataclasses import dataclass, field @@ -181,6 +182,11 @@ def __iter__(self) -> Iterator[bytes]: def __next__(self) -> bytes: try: return next(self._iterator) + except TimeoutError as e: + # The transport's idle read timeout is the builtin TimeoutError + # under pyqwest; keep the documented httpx exception. + self.close() + raise httpx.ReadTimeout(str(e)) from e except BaseException: # Covers normal end (StopIteration) and read errors alike. self.close() @@ -219,9 +225,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 +239,17 @@ def __aiter__(self) -> 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 (TimeoutError, asyncio.TimeoutError) as e: + # Both the transport's idle read timeout (the builtin + # TimeoutError under pyqwest) and wait_for's expiry (a distinct + # asyncio.TimeoutError until 3.11); keep the documented httpx + # exception. + await self.aclose() + raise httpx.ReadTimeout(str(e)) from e except BaseException: # Covers normal end (StopAsyncIteration) and read errors alike. await self.aclose() diff --git a/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py b/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py index 24d25d9330..f1bd61ea3a 100644 --- a/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py +++ b/packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py @@ -7,6 +7,7 @@ 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, @@ -93,6 +94,11 @@ def __init__( 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 +163,26 @@ async def read( """ 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 +213,53 @@ async def read( 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 (TimeoutError, 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( @@ -351,9 +377,12 @@ async def write_files( # 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 +404,14 @@ async def _upload_file(file): 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 +454,9 @@ async def _upload_file(file): files=httpx_files, params=params, headers=extra_headers, - timeout=upload_timeout, + # A multipart body with a file-like entry is streamed too + # (httpx forwards `IOBase` entries in chunks). + timeout=None if has_streamable_data else upload_timeout, ) except httpx.RemoteProtocolError as e: raise await ahandle_envd_api_transport_exception_with_health( diff --git a/packages/python-sdk/e2b/sandbox_async/main.py b/packages/python-sdk/e2b/sandbox_async/main.py index 9af0935bf0..0731077e37 100644 --- a/packages/python-sdk/e2b/sandbox_async/main.py +++ b/packages/python-sdk/e2b/sandbox_async/main.py @@ -9,9 +9,8 @@ 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 @@ -105,13 +104,7 @@ def __init__( """ 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, diff --git a/packages/python-sdk/e2b/sandbox_sync/commands/command.py b/packages/python-sdk/e2b/sandbox_sync/commands/command.py index d1cf6e0fe3..a606105191 100644 --- a/packages/python-sdk/e2b/sandbox_sync/commands/command.py +++ b/packages/python-sdk/e2b/sandbox_sync/commands/command.py @@ -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 @@ def __init__( 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) diff --git a/packages/python-sdk/e2b/sandbox_sync/commands/pty.py b/packages/python-sdk/e2b/sandbox_sync/commands/pty.py index 90f2ee4d10..34b32cdc24 100644 --- a/packages/python-sdk/e2b/sandbox_sync/commands/pty.py +++ b/packages/python-sdk/e2b/sandbox_sync/commands/pty.py @@ -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 @@ def __init__( 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) diff --git a/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py b/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py index aaecb07cf1..25ccf071f2 100644 --- a/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py +++ b/packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py @@ -1,4 +1,3 @@ -import threading from typing import IO, Dict, List, Literal, Optional, Union, overload import httpx @@ -6,8 +5,7 @@ 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, @@ -88,32 +86,19 @@ def __init__( 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 +156,23 @@ def read( """ 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 +203,24 @@ def read( 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 +229,6 @@ def read( 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: @@ -359,9 +345,12 @@ def write_files( # 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 +371,14 @@ def write_files( 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 +415,9 @@ def write_files( files=httpx_files, params=params, headers=extra_headers, - timeout=upload_timeout, + # A multipart body with a file-like entry is streamed too + # (httpx forwards `IOBase` entries in chunks). + timeout=None if has_streamable_data else upload_timeout, ) except httpx.RemoteProtocolError as e: raise handle_envd_api_transport_exception_with_health(e, self._envd_api) diff --git a/packages/python-sdk/tests/async/sandbox_async/test_config_propagation.py b/packages/python-sdk/tests/async/sandbox_async/test_config_propagation.py index 1acfe87800..feada7d5ef 100644 --- a/packages/python-sdk/tests/async/sandbox_async/test_config_propagation.py +++ b/packages/python-sdk/tests/async/sandbox_async/test_config_propagation.py @@ -1,4 +1,3 @@ -from types import SimpleNamespace from unittest.mock import AsyncMock import pytest @@ -16,13 +15,8 @@ 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() diff --git a/packages/python-sdk/tests/sync/sandbox_sync/test_sync_client_lifecycle.py b/packages/python-sdk/tests/sync/sandbox_sync/test_sync_client_lifecycle.py index f8b0ddba02..cdbe42a758 100644 --- a/packages/python-sdk/tests/sync/sandbox_sync/test_sync_client_lifecycle.py +++ b/packages/python-sdk/tests/sync/sandbox_sync/test_sync_client_lifecycle.py @@ -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 diff --git a/packages/python-sdk/tests/test_api_client_transport.py b/packages/python-sdk/tests/test_api_client_transport.py index 97fc9c32fd..4700511712 100644 --- a/packages/python-sdk/tests/test_api_client_transport.py +++ b/packages/python-sdk/tests/test_api_client_transport.py @@ -2,52 +2,37 @@ 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 ( - Headers, - Proxy, - Request, - Response, - SyncRequest, - SyncResponse, - SyncTransport, - Transport, -) +from pyqwest import Headers, Response, SyncResponse 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): @@ -55,76 +40,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_transport_strips_host_header(): # Forwarding httpx's auto-added Host header on an HTTP/2 connection makes # the E2B API edge reset the stream with PROTOCOL_ERROR; hyper derives @@ -240,26 +155,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): @@ -281,23 +218,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() @@ -368,24 +288,41 @@ async def get_client(): @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() + + +@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() - AsyncEnvdTransportWithLogger._instances.clear() class _EchoHandler(BaseHTTPRequestHandler): @@ -403,6 +340,16 @@ def do_GET(self): self.end_headers() 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 @@ -580,3 +527,24 @@ async def test_async_api_client_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() diff --git a/packages/python-sdk/tests/test_env_var_parsing.py b/packages/python-sdk/tests/test_env_var_parsing.py index 8089f16c4a..9b2b0edaf2 100644 --- a/packages/python-sdk/tests/test_env_var_parsing.py +++ b/packages/python-sdk/tests/test_env_var_parsing.py @@ -5,7 +5,6 @@ import importlib import e2b.api -import e2b.envd.client_shared _ENV_VARS = ( "E2B_KEEPALIVE_EXPIRY", @@ -16,19 +15,16 @@ 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 @@ -42,8 +38,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() diff --git a/packages/python-sdk/tests/test_envd_client_transport.py b/packages/python-sdk/tests/test_envd_client_transport.py index d4f72cb500..683d65faea 100644 --- a/packages/python-sdk/tests/test_envd_client_transport.py +++ b/packages/python-sdk/tests/test_envd_client_transport.py @@ -2,10 +2,13 @@ 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 @@ -18,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(cast(ProxyTypes, 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 @@ -101,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 diff --git a/packages/python-sdk/tests/test_envd_plain_http_errors.py b/packages/python-sdk/tests/test_envd_plain_http_errors.py index 7962fbd3e5..8367cce0a7 100644 --- a/packages/python-sdk/tests/test_envd_plain_http_errors.py +++ b/packages/python-sdk/tests/test_envd_plain_http_errors.py @@ -31,12 +31,12 @@ ) import e2b.sandbox_async.commands.command as command_async -from e2b.envd.client_async import ( - ConnectionRetryTransport, - PlainHTTPErrorTransport, +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 ( - ConnectionRetryTransport as SyncConnectionRetryTransport, PlainHTTPErrorTransport as SyncPlainHTTPErrorTransport, ) from e2b.envd.process.process_pb import ConnectRequest diff --git a/packages/python-sdk/tests/test_envd_retry_transport.py b/packages/python-sdk/tests/test_envd_retry_transport.py index 00f20cb661..7b1aa3172a 100644 --- a/packages/python-sdk/tests/test_envd_retry_transport.py +++ b/packages/python-sdk/tests/test_envd_retry_transport.py @@ -24,8 +24,8 @@ 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 diff --git a/packages/python-sdk/tests/test_file_stream_reader.py b/packages/python-sdk/tests/test_file_stream_reader.py index d499aec633..c3d063fb79 100644 --- a/packages/python-sdk/tests/test_file_stream_reader.py +++ b/packages/python-sdk/tests/test_file_stream_reader.py @@ -224,3 +224,71 @@ async def test_async_abandoned_reader_is_reclaimed_on_client_close(): if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-v"])) + + +class _TimeoutingResponse: + """Stands in for an httpx.Response whose body read hits the pyqwest + transport's idle read timeout (the builtin TimeoutError).""" + + def __init__(self): + self.closed = False + + def iter_bytes(self): + yield b"first" + raise TimeoutError("read timed out") + + async def aiter_bytes(self): + yield b"first" + raise TimeoutError("read timed out") + + def close(self): + self.closed = True + + async def aclose(self): + self.closed = True + + +def test_sync_reader_maps_transport_timeout_to_read_timeout(): + response = _TimeoutingResponse() + reader = FileStreamReader(response) # ty: ignore[invalid-argument-type] + it = iter(reader) + assert next(it) == b"first" + with pytest.raises(httpx.ReadTimeout): + next(it) + assert response.closed + + +async def test_async_reader_maps_transport_timeout_to_read_timeout(): + response = _TimeoutingResponse() + reader = AsyncFileStreamReader(response) # ty: ignore[invalid-argument-type] + assert await reader.__anext__() == b"first" + with pytest.raises(httpx.ReadTimeout): + await reader.__anext__() + assert response.closed + + +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