diff --git a/README.md b/README.md index c0f5c7f2f1..1b3824da05 100644 --- a/README.md +++ b/README.md @@ -657,6 +657,14 @@ response), a subclass of `openai.APIStatusError` is raised, containing `status_c All errors inherit from `openai.APIError`. +When consuming a `Stream` or `AsyncStream`, read timeouts raise `APITimeoutError` +and other HTTPX request failures raise `APIConnectionError`. Catch these SDK +exceptions instead of raw HTTPX exceptions; the original exception is available +as `__cause__`. Stream consumption is not automatically retried, because replaying +a request could duplicate output already delivered to your application. +The Assistants event-handler helpers and raw `with_streaming_response` iterators +retain their existing exception behavior. + ```python import openai from openai import OpenAI diff --git a/src/openai/_httpx2.py b/src/openai/_httpx2.py index 491398b43c..8944928d84 100644 --- a/src/openai/_httpx2.py +++ b/src/openai/_httpx2.py @@ -17,6 +17,7 @@ class _LegacyHttpxModule(Protocol): Timeout: type[httpx2.Timeout] Limits: type[httpx2.Limits] TimeoutException: type[httpx2.TimeoutException] + RequestError: type[httpx2.RequestError] HTTPStatusError: type[httpx2.HTTPStatusError] StreamConsumed: type[httpx2.StreamConsumed] RequestNotRead: type[httpx2.RequestNotRead] @@ -99,6 +100,11 @@ def timeout_exceptions() -> tuple[type[httpx2.TimeoutException], ...]: return (httpx2.TimeoutException,) if module is None else (httpx2.TimeoutException, module.TimeoutException) +def request_exceptions() -> tuple[type[httpx2.RequestError], ...]: + module = _loaded_legacy_httpx() + return (httpx2.RequestError,) if module is None else (httpx2.RequestError, module.RequestError) + + 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 0fd375604d..bb45a70f6a 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 @@ -51,7 +52,12 @@ def __iter__(self) -> Iterator[_T]: yield item def _iter_events(self) -> Iterator[ServerSentEvent]: - yield from self._decoder.iter_bytes(self.response.iter_bytes()) + try: + yield from self._decoder.iter_bytes(self.response.iter_bytes()) + except timeout_exceptions() as err: + raise APITimeoutError(request=self.response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=self.response.request) from err def __stream__(self) -> Iterator[_T]: cast_to = cast(Any, self._cast_to) @@ -160,8 +166,13 @@ async def __aiter__(self) -> AsyncIterator[_T]: yield item async def _iter_events(self) -> AsyncIterator[ServerSentEvent]: - async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()): - yield sse + try: + async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()): + yield sse + except timeout_exceptions() as err: + raise APITimeoutError(request=self.response.request) from err + except request_exceptions() as err: + raise APIConnectionError(request=self.response.request) from err async def __stream__(self) -> AsyncIterator[_T]: cast_to = cast(Any, self._cast_to) diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 1cdca954a5..676ffe6383 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -8,10 +8,11 @@ from ._deltas import accumulate_delta as accumulate_delta from ..._utils import 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 APIConnectionError from ...types.beta.threads import ( Run, Text, @@ -29,6 +30,26 @@ def _timeout_exceptions() -> tuple[type[Exception], ...]: return (*timeout_exceptions(), asyncio.TimeoutError) +def _iter_events(stream: Stream[AssistantStreamEvent]) -> Iterator[AssistantStreamEvent]: + # Preserve legacy transport exceptions without unwrapping errors from user callbacks. + try: + yield from stream + except APIConnectionError as exc: + if isinstance(exc.__cause__, request_exceptions()): + raise exc.__cause__ from None + raise + + +async def _aiter_events(stream: AsyncStream[AssistantStreamEvent]) -> AsyncIterator[AssistantStreamEvent]: + try: + async for event in stream: + yield event + except APIConnectionError as exc: + if isinstance(exc.__cause__, request_exceptions()): + raise exc.__cause__ from None + raise + + class AssistantEventHandler: text_deltas: Iterable[str] """Iterator over just the text deltas in the stream. @@ -407,7 +428,7 @@ def __stream__(self) -> Iterator[AssistantStreamEvent]: raise RuntimeError("Stream has not been started yet") try: - for event in stream: + for event in _iter_events(stream): self._emit_sse_event(event) yield event @@ -839,7 +860,7 @@ async def __stream__(self) -> AsyncIterator[AssistantStreamEvent]: raise RuntimeError("Stream has not been started yet") try: - async for event in stream: + async for event in _aiter_events(stream): await self._emit_sse_event(event) yield event diff --git a/tests/test_httpx2.py b/tests/test_httpx2.py index 764fc00f0b..bcb1036cf7 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: @@ -577,7 +578,8 @@ async def async_failing_stream_handler(request: httpx2.Request) -> httpx2.Respon @pytest.mark.filterwarnings("ignore:The Assistants API is deprecated in favor of the Responses API:DeprecationWarning") -async def test_assistant_stream_timeout_callbacks_preserve_httpx2_family() -> None: +@pytest.mark.parametrize("error_type", [httpx2.ReadTimeout, httpx2.RemoteProtocolError, httpx2.DecodingError]) +async def test_assistant_stream_error_callbacks_preserve_httpx2_family(error_type: type[httpx2.RequestError]) -> None: class SyncHandler(openai.AssistantEventHandler): def __init__(self) -> None: super().__init__() @@ -610,13 +612,13 @@ class FailingSyncStream(httpx2.SyncByteStream): @override def __iter__(self): yield b"partial" - raise httpx2.ReadTimeout("assistant stream timeout") + raise error_type("assistant stream failure") class FailingAsyncStream(httpx2.AsyncByteStream): @override async def __aiter__(self): yield b"partial" - raise httpx2.ReadTimeout("assistant stream timeout") + raise error_type("assistant stream failure") def sync_response(request: httpx2.Request) -> httpx2.Response: return httpx2.Response( @@ -638,11 +640,11 @@ async def async_response(request: httpx2.Request) -> httpx2.Response: with sync_client.beta.threads.runs.stream( # pyright: ignore[reportDeprecated] assistant_id="asst_test", thread_id="thread_test", event_handler=sync_handler ) as stream: - with pytest.raises(httpx2.ReadTimeout, match="assistant stream timeout"): + with pytest.raises(error_type, match="assistant stream failure"): stream.until_done() - assert sync_handler.timed_out - assert isinstance(sync_handler.exception, httpx2.ReadTimeout) + assert sync_handler.timed_out == issubclass(error_type, httpx2.TimeoutException) + assert isinstance(sync_handler.exception, error_type) async_handler = AsyncHandler() async with AsyncOpenAI( @@ -654,11 +656,132 @@ async def async_response(request: httpx2.Request) -> httpx2.Response: async with async_client.beta.threads.runs.stream( # pyright: ignore[reportDeprecated] assistant_id="asst_test", thread_id="thread_test", event_handler=async_handler ) as async_stream: - with pytest.raises(httpx2.ReadTimeout, match="assistant stream timeout"): + with pytest.raises(error_type, match="assistant stream failure"): await async_stream.until_done() - assert async_handler.timed_out - assert isinstance(async_handler.exception, httpx2.ReadTimeout) + assert async_handler.timed_out == issubclass(error_type, httpx2.TimeoutException) + assert isinstance(async_handler.exception, error_type) + + +@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: diff --git a/tests/test_streaming.py b/tests/test_streaming.py index ee3af47ab6..fe74fcc558 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -1,15 +1,160 @@ from __future__ import annotations -from typing import Iterator, AsyncIterator +import os +import importlib +from typing import Any, Iterator, AsyncIterator from contextlib import aclosing, nullcontext import httpx2 import pytest -from openai import OpenAI, AsyncOpenAI +from openai import OpenAI, AsyncOpenAI, APITimeoutError, APIConnectionError from openai._streaming import Stream, AsyncStream, ServerSentEvent +@pytest.fixture( + params=[ + "httpx2", + pytest.param( + "httpx", + marks=pytest.mark.skipif( + os.environ.get("OPENAI_TEST_LEGACY_HTTPX") != "1", reason="requires the legacy HTTPX compatibility lane" + ), + ), + ] +) +def http_module(request: pytest.FixtureRequest) -> Any: + return importlib.import_module(request.param) + + +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize("delivered", [False, True], ids=["before-first-event", "after-first-event"]) +@pytest.mark.parametrize( + ("error_name", "expected_error"), + [ + ("ReadTimeout", APITimeoutError), + ("RemoteProtocolError", APIConnectionError), + ("DecodingError", APIConnectionError), + ], +) +async def test_request_errors_are_wrapped( + sync: bool, delivered: bool, error_name: str, expected_error: type[APIConnectionError], http_module: Any +) -> None: + error = getattr(http_module, error_name)("synthetic stream failure") + requests: list[Any] = [] + first = ( + b'data: {"id":"synthetic","object":"chat.completion.chunk","created":0,"model":"synthetic",' + b'"choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}\n\n' + ) + + def body() -> Iterator[bytes]: + if delivered: + yield first + raise error + + async def async_body() -> AsyncIterator[bytes]: + for chunk in body(): + yield chunk + + def handler(request: Any) -> Any: + requests.append(request) + return http_module.Response( + 200, headers={"content-type": "text/event-stream"}, content=body() if sync else async_body() + ) + + received: list[str | None] = [] + if sync: + with OpenAI( + api_key="synthetic", + max_retries=2, + http_client=http_module.Client(transport=http_module.MockTransport(handler), trust_env=False), + ) as client: + stream = client.chat.completions.create(model="synthetic", messages=[], stream=True) + with pytest.raises(expected_error) as caught: + for chunk in stream: + received.append(chunk.choices[0].delta.content) + assert stream.response.is_closed + else: + async with AsyncOpenAI( + api_key="synthetic", + max_retries=2, + http_client=http_module.AsyncClient(transport=http_module.MockTransport(handler), trust_env=False), + ) as async_client: + async_stream = await async_client.chat.completions.create(model="synthetic", messages=[], stream=True) + with pytest.raises(expected_error) as caught: + async for chunk in async_stream: + received.append(chunk.choices[0].delta.content) + assert async_stream.response.is_closed + + assert received == (["hello"] if delivered else []) + assert len(requests) == 1 + assert caught.value.request is requests[0] + assert caught.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, + ValueError, + ], +) +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: