From f62ba8948f74fa01cf84cf4ef42a7276df85615b Mon Sep 17 00:00:00 2001 From: heri-espino Date: Wed, 9 Sep 2026 06:47:37 -0600 Subject: [PATCH 1/3] fix(streaming): wrap mid-stream request errors --- src/openai/_httpx2.py | 12 ++++ src/openai/_streaming.py | 11 +++- src/openai/lib/streaming/_assistants.py | 45 ++++++++++++++- tests/test_streaming.py | 75 ++++++++++++++++++++++++- 4 files changed, 140 insertions(+), 3 deletions(-) diff --git a/src/openai/_httpx2.py b/src/openai/_httpx2.py index 491398b43c..9655ad6f27 100644 --- a/src/openai/_httpx2.py +++ b/src/openai/_httpx2.py @@ -99,6 +99,18 @@ def timeout_exceptions() -> tuple[type[httpx2.TimeoutException], ...]: return (httpx2.TimeoutException,) if module is None else (httpx2.TimeoutException, module.TimeoutException) +def request_exceptions() -> tuple[type[Exception], ...]: + module = _loaded_legacy_httpx() + if module is None: + return (httpx2.RequestError,) + + legacy_request_error = cast( + type[Exception], + getattr(module, "RequestError", httpx2.RequestError), + ) + return (httpx2.RequestError, legacy_request_error) + + def status_exceptions() -> tuple[type[httpx2.HTTPStatusError], ...]: module = _loaded_legacy_httpx() return (httpx2.HTTPStatusError,) if module is None else (httpx2.HTTPStatusError, module.HTTPStatusError) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 78e2d20aa7..3360980a5c 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -11,7 +11,8 @@ import httpx2 from ._utils import is_mapping, extract_type_var_from_base -from ._exceptions import APIError +from ._httpx2 import request_exceptions, timeout_exceptions +from ._exceptions import APIError, APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import OpenAI, AsyncOpenAI @@ -106,6 +107,10 @@ def __stream__(self) -> Iterator[_T]: cast_to=cast_to, response=response, ) + except timeout_exceptions() as err: + raise APITimeoutError(request=response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=response.request) from err finally: # Ensure the response is closed even if the consumer doesn't read all data response.close() @@ -216,6 +221,10 @@ async def __stream__(self) -> AsyncIterator[_T]: cast_to=cast_to, response=response, ) + except timeout_exceptions() as err: + raise APITimeoutError(request=response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=response.request) from err finally: # Ensure the response is closed even if the consumer doesn't read all data await response.aclose() diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 314961230d..233bdc5cef 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -7,10 +7,11 @@ from ..._utils import is_dict, is_list, consume_sync_iterator, consume_async_iterator from ..._compat import model_dump -from ..._httpx2 import timeout_exceptions +from ..._httpx2 import request_exceptions, timeout_exceptions from ..._models import construct_type from ..._streaming import Stream, AsyncStream from ...types.beta import AssistantStreamEvent +from ..._exceptions import APITimeoutError, APIConnectionError from ...types.beta.threads import ( Run, Text, @@ -28,6 +29,14 @@ def _timeout_exceptions() -> tuple[type[Exception], ...]: return (*timeout_exceptions(), asyncio.TimeoutError) +def _request_error_from_api_error(error: APIConnectionError) -> Exception | None: + cause = error.__cause__ + if isinstance(cause, request_exceptions()): + return cause + + return None + + class AssistantEventHandler: text_deltas: Iterable[str] """Iterator over just the text deltas in the stream. @@ -410,6 +419,23 @@ def __stream__(self) -> Iterator[AssistantStreamEvent]: self._emit_sse_event(event) yield event + except APITimeoutError as exc: + error = _request_error_from_api_error(exc) + self.on_timeout() + self.on_exception(error or exc) + + if error is None: + raise + + raise error from None + except APIConnectionError as exc: + error = _request_error_from_api_error(exc) + self.on_exception(error or exc) + + if error is None: + raise + + raise error from None except _timeout_exceptions() as exc: self.on_timeout() self.on_exception(exc) @@ -842,6 +868,23 @@ async def __stream__(self) -> AsyncIterator[AssistantStreamEvent]: await self._emit_sse_event(event) yield event + except APITimeoutError as exc: + error = _request_error_from_api_error(exc) + await self.on_timeout() + await self.on_exception(error or exc) + + if error is None: + raise + + raise error from None + except APIConnectionError as exc: + error = _request_error_from_api_error(exc) + await self.on_exception(error or exc) + + if error is None: + raise + + raise error from None except _timeout_exceptions() as exc: await self.on_timeout() await self.on_exception(exc) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index ae6c0590f7..37e73899da 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -1,14 +1,87 @@ from __future__ import annotations from typing import Iterator, AsyncIterator +from typing_extensions import override import httpx2 import pytest -from openai import OpenAI, AsyncOpenAI +from openai import OpenAI, AsyncOpenAI, APITimeoutError, APIConnectionError from openai._streaming import Stream, AsyncStream, ServerSentEvent +class FailingSyncByteStream(httpx2.SyncByteStream): + def __init__(self, error: Exception) -> None: + self.error = error + + @override + def __iter__(self) -> Iterator[bytes]: + yield b'data: {"foo":' + raise self.error + + +class FailingAsyncByteStream(httpx2.AsyncByteStream): + def __init__(self, error: Exception) -> None: + self.error = error + + @override + async def __aiter__(self) -> AsyncIterator[bytes]: + yield b'data: {"foo":' + raise self.error + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize( + ("error_type", "expected_error"), + [ + (httpx2.ReadTimeout, APITimeoutError), + (httpx2.RemoteProtocolError, APIConnectionError), + (httpx2.DecodingError, APIConnectionError), + ], +) +async def test_request_errors_are_wrapped( + sync: bool, + error_type: type[Exception], + expected_error: type[Exception], + client: OpenAI, + async_client: AsyncOpenAI, +) -> None: + error = error_type("stream request failure") + request = httpx2.Request("POST", "https://example.com") + + if sync: + response = httpx2.Response( + 200, + request=request, + stream=FailingSyncByteStream(error), + ) + stream = Stream( + cast_to=object, + client=client, + response=response, + ) + + with pytest.raises(expected_error) as exc_info: + next(stream) + else: + response = httpx2.Response( + 200, + request=request, + stream=FailingAsyncByteStream(error), + ) + stream = AsyncStream( + cast_to=object, + client=async_client, + response=response, + ) + + with pytest.raises(expected_error) as exc_info: + await stream.__anext__() + + assert exc_info.value.__cause__ is error + + @pytest.mark.asyncio @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) async def test_basic(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: From 80686f36241df1cef3bc7d87f3deddca30bcf183 Mon Sep 17 00:00:00 2001 From: heri-espino Date: Wed, 9 Sep 2026 07:53:18 -0600 Subject: [PATCH 2/3] fix(streaming): limit assistant unwrapping to iterator --- src/openai/lib/streaming/_assistants.py | 62 +++++------- tests/test_httpx2.py | 124 ++++++++++++++++++++++++ 2 files changed, 149 insertions(+), 37 deletions(-) diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 233bdc5cef..db807e4ed4 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -11,7 +11,7 @@ from ..._models import construct_type from ..._streaming import Stream, AsyncStream from ...types.beta import AssistantStreamEvent -from ..._exceptions import APITimeoutError, APIConnectionError +from ..._exceptions import APIConnectionError from ...types.beta.threads import ( Run, Text, @@ -415,27 +415,21 @@ def __stream__(self) -> Iterator[AssistantStreamEvent]: raise RuntimeError("Stream has not been started yet") try: - for event in stream: - self._emit_sse_event(event) - - yield event - except APITimeoutError as exc: - error = _request_error_from_api_error(exc) - self.on_timeout() - self.on_exception(error or exc) - - if error is None: - raise + while True: + try: + event = next(stream) + except StopIteration: + break + except APIConnectionError as exc: + error = _request_error_from_api_error(exc) - raise error from None - except APIConnectionError as exc: - error = _request_error_from_api_error(exc) - self.on_exception(error or exc) + if error is None: + raise - if error is None: - raise + raise error from None - raise error from None + self._emit_sse_event(event) + yield event except _timeout_exceptions() as exc: self.on_timeout() self.on_exception(exc) @@ -864,27 +858,21 @@ async def __stream__(self) -> AsyncIterator[AssistantStreamEvent]: raise RuntimeError("Stream has not been started yet") try: - async for event in stream: - await self._emit_sse_event(event) - - yield event - except APITimeoutError as exc: - error = _request_error_from_api_error(exc) - await self.on_timeout() - await self.on_exception(error or exc) - - if error is None: - raise + while True: + try: + event = await stream.__anext__() + except StopAsyncIteration: + break + except APIConnectionError as exc: + error = _request_error_from_api_error(exc) - raise error from None - except APIConnectionError as exc: - error = _request_error_from_api_error(exc) - await self.on_exception(error or exc) + if error is None: + raise - if error is None: - raise + raise error from None - raise error from None + await self._emit_sse_event(event) + yield event except _timeout_exceptions() as exc: await self.on_timeout() await self.on_exception(exc) diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py index 764fc00f0b..4cfa98d896 100644 --- a/tests/test_httpx2.py +++ b/tests/test_httpx2.py @@ -21,6 +21,7 @@ from openai._response import StreamAlreadyConsumed from openai.providers import bedrock from openai._constants import DEFAULT_TIMEOUT +from openai.types.beta import AssistantStreamEvent def model_list(request: httpx2.Request) -> httpx2.Response: @@ -661,6 +662,129 @@ async def async_response(request: httpx2.Request) -> httpx2.Response: assert isinstance(async_handler.exception, httpx2.ReadTimeout) +@pytest.mark.filterwarnings( + "ignore:The Assistants API is deprecated in favor of the Responses API:DeprecationWarning" +) +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize("timeout", [True, False], ids=["timeout", "connection"]) +async def test_assistant_stream_hook_api_errors_are_not_unwrapped(sync: bool, timeout: bool) -> None: + hook_request = httpx2.Request("GET", "https://example.test/hook") + + if timeout: + cause: httpx2.RequestError = httpx2.ReadTimeout("hook timeout") + hook_error: APIConnectionError = APITimeoutError(hook_request) + else: + cause = httpx2.RemoteProtocolError("hook connection error") + hook_error = APIConnectionError(request=hook_request) + + hook_error.__cause__ = cause + + class SyncHandler(openai.AssistantEventHandler): + def __init__(self) -> None: + super().__init__() + self.timed_out = False + self.exception: Exception | None = None + + @override + def on_event(self, event: AssistantStreamEvent) -> None: + raise hook_error + + @override + def on_timeout(self) -> None: + self.timed_out = True + + @override + def on_exception(self, exception: Exception) -> None: + self.exception = exception + + class AsyncHandler(openai.AsyncAssistantEventHandler): + def __init__(self) -> None: + super().__init__() + self.timed_out = False + self.exception: Exception | None = None + + @override + async def on_event(self, event: AssistantStreamEvent) -> None: + raise hook_error + + @override + async def on_timeout(self) -> None: + self.timed_out = True + + @override + async def on_exception(self, exception: Exception) -> None: + self.exception = exception + + content = b'event: thread.created\ndata: {"id":"thread_test","created_at":0,"object":"thread"}\n\n' + + def sync_response(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=content, + request=request, + ) + + async def async_response(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=content, + request=request, + ) + + if sync: + handler = SyncHandler() + + with OpenAI( + api_key="test", + base_url="https://example.test/v1", + http_client=openai.DefaultHttpx2Client( + transport=httpx2.MockTransport(sync_response), + trust_env=False, + ), + max_retries=0, + ) as client: + with client.beta.threads.runs.stream( # pyright: ignore[reportDeprecated] + assistant_id="asst_test", + thread_id="thread_test", + event_handler=handler, + ) as stream: + with pytest.raises(APIConnectionError) as exc_info: + stream.until_done() + + assert exc_info.value is hook_error + assert handler.exception is hook_error + assert not handler.timed_out + assert hook_error.__cause__ is cause + + else: + handler = AsyncHandler() + + async with AsyncOpenAI( + api_key="test", + base_url="https://example.test/v1", + http_client=openai.DefaultAsyncHttpx2Client( + transport=httpx2.MockTransport(async_response), + trust_env=False, + ), + max_retries=0, + ) as client: + async with client.beta.threads.runs.stream( # pyright: ignore[reportDeprecated] + assistant_id="asst_test", + thread_id="thread_test", + event_handler=handler, + ) as stream: + with pytest.raises(APIConnectionError) as exc_info: + await stream.until_done() + + assert exc_info.value is hook_error + assert handler.exception is hook_error + assert not handler.timed_out + assert hook_error.__cause__ is cause + + async def test_sigv4_provider_preserves_httpx2_family_and_rejects_one_shot_bodies() -> None: pytest.importorskip("botocore") From 1be10fc66338b2aea0f006a5492dd255c47c6478 Mon Sep 17 00:00:00 2001 From: heri-espino Date: Wed, 9 Sep 2026 08:26:14 -0600 Subject: [PATCH 3/3] fix(streaming): scope request error wrapping to iteration --- src/openai/_streaming.py | 28 +++++++++++------- tests/test_streaming.py | 61 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 10 deletions(-) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 3360980a5c..2b869f8ed1 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -61,7 +61,15 @@ def __stream__(self) -> Iterator[_T]: iterator = self._iter_events() try: - for sse in iterator: + while True: + try: + sse = next(iterator) + except StopIteration: + break + except timeout_exceptions() as err: + raise APITimeoutError(request=response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=response.request) from err if sse.data.startswith("[DONE]"): break @@ -107,10 +115,6 @@ def __stream__(self) -> Iterator[_T]: cast_to=cast_to, response=response, ) - except timeout_exceptions() as err: - raise APITimeoutError(request=response.request) from err - except request_exceptions() as err: - raise APIConnectionError(request=response.request) from err finally: # Ensure the response is closed even if the consumer doesn't read all data response.close() @@ -175,7 +179,15 @@ async def __stream__(self) -> AsyncIterator[_T]: iterator = self._iter_events() try: - async for sse in iterator: + while True: + try: + sse = await iterator.__anext__() + except StopAsyncIteration: + break + except timeout_exceptions() as err: + raise APITimeoutError(request=response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=response.request) from err if sse.data.startswith("[DONE]"): break @@ -221,10 +233,6 @@ async def __stream__(self) -> AsyncIterator[_T]: cast_to=cast_to, response=response, ) - except timeout_exceptions() as err: - raise APITimeoutError(request=response.request) from err - except request_exceptions() as err: - raise APIConnectionError(request=response.request) from err finally: # Ensure the response is closed even if the consumer doesn't read all data await response.aclose() diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 37e73899da..d03099a95d 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -82,6 +82,67 @@ async def test_request_errors_are_wrapped( assert exc_info.value.__cause__ is error +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize( + "error_type", + [ + httpx2.ReadTimeout, + httpx2.RemoteProtocolError, + ], +) +async def test_request_errors_from_response_processing_are_not_wrapped( + sync: bool, + error_type: type[Exception], + client: OpenAI, + async_client: AsyncOpenAI, +) -> None: + error = error_type("response processing failure") + request = httpx2.Request("POST", "https://example.com") + + class FailingModelBuilder: + @classmethod + def build( + cls, + *, + response: httpx2.Response, + data: object, + ) -> FailingModelBuilder: + assert response.request is request + assert data == {"foo": True} + raise error + + if sync: + response = httpx2.Response( + 200, + request=request, + content=b'data: {"foo": true}\n\n', + ) + stream = Stream( + cast_to=FailingModelBuilder, + client=client, + response=response, + ) + + with pytest.raises(error_type) as exc_info: + next(stream) + else: + response = httpx2.Response( + 200, + request=request, + content=b'data: {"foo": true}\n\n', + ) + stream = AsyncStream( + cast_to=FailingModelBuilder, + client=async_client, + response=response, + ) + + with pytest.raises(error_type) as exc_info: + await stream.__anext__() + + assert exc_info.value is error + @pytest.mark.asyncio @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) async def test_basic(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: