diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index ae718ce10..0364140c4 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -13,6 +13,7 @@ from ._utils import extract_type_var_from_base from ._compat import cached_property +from ._exceptions import APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -68,7 +69,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 httpx2.TimeoutException as err: + raise APITimeoutError(request=self.response.request) from err + except httpx2.TransportError as err: + raise APIConnectionError(request=self.response.request) from err @staticmethod def raw_events(response: httpx2.Response) -> Iterator[ServerSentEvent]: @@ -196,8 +202,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 httpx2.TimeoutException as err: + raise APITimeoutError(request=self.response.request) from err + except httpx2.TransportError as err: + raise APIConnectionError(request=self.response.request) from err @staticmethod def raw_events(response: httpx2.Response) -> AsyncIterator[ServerSentEvent]: diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 5efd07a67..3b0b18acc 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -8,7 +8,7 @@ from anthropic import Anthropic, AsyncAnthropic from anthropic._streaming import Stream, AsyncStream, ServerSentEvent -from anthropic._exceptions import APIStatusError +from anthropic._exceptions import APIStatusError, APITimeoutError, APIConnectionError from .utils import ( STREAMS, @@ -348,6 +348,64 @@ def body() -> Iterator[bytes]: assert "Overloaded" in str(exc_info.value) +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_mid_stream_timeout_wrapped_as_api_timeout_error( + sync: bool, client: Anthropic, async_client: AsyncAnthropic +) -> None: + def body() -> Iterator[bytes]: + yield b"event: completion\n" + yield b'data: {"foo":true}\n' + yield b"\n" + raise httpx2.ReadTimeout("timed out while reading the stream") + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + # the first event still comes through fine + assert await iter_next(iterator) == {"foo": True} + + with pytest.raises(APITimeoutError): + await iter_next(iterator) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_mid_stream_transport_error_wrapped_as_api_connection_error( + sync: bool, client: Anthropic, async_client: AsyncAnthropic +) -> None: + def body() -> Iterator[bytes]: + yield b"event: completion\n" + yield b'data: {"foo":true}\n' + yield b"\n" + raise httpx2.RemoteProtocolError("connection dropped mid-stream") + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + assert await iter_next(iterator) == {"foo": True} + + with pytest.raises(APIConnectionError): + await iter_next(iterator) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_mid_stream_decoder_error_keeps_original_identity( + sync: bool, client: Anthropic, async_client: AsyncAnthropic +) -> None: + # malformed UTF-8 in an SSE line: the default decoder raises + # UnicodeDecodeError, which is not a transport failure and must not + # be relabeled as APIConnectionError. + def body() -> Iterator[bytes]: + yield b"event: completion\n" + yield b"data: \xff\xfe not utf-8\n" + yield b"\n" + + iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(UnicodeDecodeError): + await iter_next(iterator) + + def test_isinstance_check(client: Anthropic, async_client: AsyncAnthropic) -> None: async_stream = AsyncStream(cast_to=object, client=async_client, response=httpx2.Response(200, content=b"foo")) assert isinstance(async_stream, AsyncStream)