diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 0b98f410c..1dfc5182d 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -10,12 +10,47 @@ import httpx2 from ._utils import is_dict, extract_type_var_from_base +from ._exceptions import ( + NotFoundError, + APIStatusError, + RateLimitError, + BadRequestError, + OverloadedError, + AuthenticationError, + InternalServerError, + PermissionDeniedError, +) if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic from ._models import FinalRequestOptions +# An error event inside a stream arrives on the stream's own HTTP response, whose status is 200, +# so the status code cannot pick the exception class the way it does for a failed request. +# The error type in the body can; the status code and the response stay the real ones. +_STREAM_ERROR_TYPE_TO_CLASS: dict[str, type[APIStatusError]] = { + "invalid_request_error": BadRequestError, + "authentication_error": AuthenticationError, + "permission_error": PermissionDeniedError, + "not_found_error": NotFoundError, + "rate_limit_error": RateLimitError, + "overloaded_error": OverloadedError, + "api_error": InternalServerError, +} + + +def _make_stream_status_error( + client: Anthropic | AsyncAnthropic, err_msg: str, *, body: object, response: httpx2.Response +) -> APIStatusError: + error = body.get("error") if is_dict(body) else None + error_type = error.get("type") if is_dict(error) else None + error_class = _STREAM_ERROR_TYPE_TO_CLASS.get(error_type) if isinstance(error_type, str) else None + if error_class is not None: + return error_class(err_msg, response=response, body=body) + return client._make_status_error(err_msg, body=body, response=response) + + _T = TypeVar("_T") @@ -137,7 +172,8 @@ def __stream__(self) -> Iterator[_T]: except Exception: err_msg = sse.data or f"Error code: {response.status_code}" - raise self._client._make_status_error( + raise _make_stream_status_error( + self._client, err_msg, body=body, response=self.response, @@ -285,7 +321,8 @@ async def __stream__(self) -> AsyncIterator[_T]: except Exception: err_msg = sse.data or f"Error code: {response.status_code}" - raise self._client._make_status_error( + raise _make_stream_status_error( + self._client, err_msg, body=body, response=self.response, diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 42106c645..df4ab6266 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -7,7 +7,16 @@ from anthropic import Anthropic, AsyncAnthropic from anthropic._streaming import Stream, AsyncStream, ServerSentEvent -from anthropic._exceptions import APIStatusError +from anthropic._exceptions import ( + NotFoundError, + APIStatusError, + RateLimitError, + BadRequestError, + OverloadedError, + AuthenticationError, + InternalServerError, + PermissionDeniedError, +) _T = TypeVar("_T") @@ -238,6 +247,60 @@ def body() -> Iterator[bytes]: assert "Overloaded" in str(exc_info.value) +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize( + "error_type,error_class", + [ + ("invalid_request_error", BadRequestError), + ("authentication_error", AuthenticationError), + ("permission_error", PermissionDeniedError), + ("not_found_error", NotFoundError), + ("rate_limit_error", RateLimitError), + ("overloaded_error", OverloadedError), + ("api_error", InternalServerError), + ], +) +async def test_error_class_from_type( + sync: bool, + error_type: str, + error_class: type[APIStatusError], + client: Anthropic, + async_client: AsyncAnthropic, +) -> None: + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield f'data: {{"type": "error", "error": {{"type": "{error_type}", "message": "boom"}}}}\n\n'.encode() + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(error_class) as exc_info: + await iter_next(iterator) + + # the response is the stream's own, so its status is reported as is + assert exc_info.value.status_code == 200 + assert exc_info.value.type == error_type + assert "boom" in str(exc_info.value) + + +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_error_of_unknown_type_stays_generic( + sync: bool, + client: Anthropic, + async_client: AsyncAnthropic, +) -> None: + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield b'data: {"type": "error", "error": {"type": "billing_error", "message": "boom"}}\n\n' + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(APIStatusError) as exc_info: + await iter_next(iterator) + + assert type(exc_info.value) is APIStatusError + assert exc_info.value.type == "billing_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)