diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 0b98f410c..e8e3e7cb9 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -10,6 +10,7 @@ import httpx2 from ._utils import is_dict, extract_type_var_from_base +from ._exceptions import APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -49,7 +50,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 42106c645..36c341094 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -7,7 +7,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 _T = TypeVar("_T") @@ -238,6 +238,37 @@ def body() -> Iterator[bytes]: assert "Overloaded" in str(exc_info.value) +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize( + ("transport_error", "api_error"), + [ + (httpx2.ReadTimeout, APITimeoutError), + (httpx2.RemoteProtocolError, APIConnectionError), + ], + ids=["timeout", "connection"], +) +async def test_transport_error_type( + sync: bool, + transport_error: type[Exception], + api_error: type[APIConnectionError], + client: Anthropic, + async_client: AsyncAnthropic, +) -> None: + def body() -> Iterator[bytes]: + yield b"event: completion\n" + yield b'data: {"foo":true}\n\n' + raise transport_error("stream interrupted") + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + assert await iter_next(iterator) == {"foo": True} + + with pytest.raises(api_error) as exc_info: + await iter_next(iterator) + + assert isinstance(exc_info.value.__cause__, transport_error) + + 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)