fix(streaming): raise the exception class matching an in-stream error's type

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,

tests/test_streaming.py

@@ -1,13 +1,23 @@
 from __future__ import annotations
 
-from typing import TypeVar, Iterator, AsyncIterator
+from typing import TypeVar, Iterator, AsyncIterator, get_args
 
 import httpx2
 import pytest
 
 from anthropic import Anthropic, AsyncAnthropic
-from anthropic._streaming import Stream, AsyncStream, ServerSentEvent
-from anthropic._exceptions import APIStatusError
+from anthropic.types import ErrorType
+from anthropic._streaming import _STREAM_ERROR_TYPE_TO_CLASS, Stream, AsyncStream, ServerSentEvent
+from anthropic._exceptions import (
+    NotFoundError,
+    APIStatusError,
+    RateLimitError,
+    BadRequestError,
+    OverloadedError,
+    AuthenticationError,
+    InternalServerError,
+    PermissionDeniedError,
+)
 
 _T = TypeVar("_T")
 
@@ -238,6 +248,66 @@ 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_every_error_type_is_mapped_or_deliberately_unmapped() -> None:
+    # no client has a class for billing_error, and timeout_error's class differs by client
+    deliberately_unmapped = {"billing_error", "timeout_error"}
+    assert set(get_args(ErrorType)) == set(_STREAM_ERROR_TYPE_TO_CLASS) | deliberately_unmapped
+
+
 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)