Skip to content

fix(streaming): wrap mid-stream transport errors as APITimeoutError/APIConnectionError #3818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adhavan18 wants to merge 1 commit into
base: main
Choose a base branch
from
+66 −2
Open

fix(streaming): wrap mid-stream transport errors as APITimeoutError/APIConnectionError #3818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion src/openai/_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@
import httpx2

from ._utils import is_mapping, extract_type_var_from_base
from ._exceptions import APIError
from ._httpx2 import timeout_exceptions
from ._exceptions import APIError, OpenAIError, APITimeoutError, APIConnectionError

if TYPE_CHECKING:
from ._client import OpenAI, AsyncOpenAI
Expand Down Expand Up @@ -106,6 +107,12 @@ def __stream__(self) -> Iterator[_T]:
cast_to=cast_to,
response=response,
)
except timeout_exceptions() as err:
raise APITimeoutError(request=response.request) from err
Comment on lines +110 to +111

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve timeout callbacks for Assistant streams

When an Assistants stream raises httpx2.ReadTimeout, this conversion means AssistantEventHandler.__stream__ and its async counterpart no longer match their _timeout_exceptions() branches, so they invoke only on_exception() and skip on_timeout(). This also changes the exception exposed by until_done() and breaks the existing sync and async callback expectations in tests/test_httpx2.py::test_assistant_stream_timeout_callbacks_preserve_httpx2_family; recognize APITimeoutError in those handlers (or preserve the transport exception for that path).

Useful? React with 👍 / 👎.

except OpenAIError:
raise
except Exception as err:
raise APIConnectionError(request=response.request) from err
Comment on lines +114 to +115

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Limit connection wrapping to transport errors

This catch also converts exceptions unrelated to the connection into APIConnectionError: malformed SSE data can raise UnicodeDecodeError or JSONDecodeError, and custom stream implementations can raise ordinary application exceptions. That changes their semantics and already makes the sync and async interruption checks in tests/test_sse_framing.py fail because the expected RuntimeError is replaced; restrict this branch to the HTTPX/HTTPX2 transport-error families so only the intended mid-stream failures are wrapped.

Useful? React with 👍 / 👎.

finally:
# Ensure the response is closed even if the consumer doesn't read all data
response.close()
Expand Down Expand Up @@ -216,6 +223,12 @@ async def __stream__(self) -> AsyncIterator[_T]:
cast_to=cast_to,
response=response,
)
except timeout_exceptions() as err:
raise APITimeoutError(request=response.request) from err
except OpenAIError:
raise
except Exception as err:
raise APIConnectionError(request=response.request) from err
finally:
# Ensure the response is closed even if the consumer doesn't read all data
await response.aclose()
Expand Down
53 changes: 52 additions & 1 deletion tests/test_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,60 @@
import httpx2
import pytest

from openai import OpenAI, AsyncOpenAI
from openai import OpenAI, AsyncOpenAI, APITimeoutError
from openai._streaming import Stream, AsyncStream, ServerSentEvent

FIRST_CHUNK = (
b'data: {"id":"c1","object":"chat.completion.chunk","created":0,"model":"gpt-5.2",'
b'"choices":[{"index":0,"delta":{"role":"assistant","content":"hi"},"finish_reason":null}]}\n\n'
)


class _DiesMidStream(httpx2.SyncByteStream):
def __iter__(self) -> Iterator[bytes]:
yield FIRST_CHUNK
raise httpx2.ReadTimeout("timed out while reading the stream")


class _AsyncDiesMidStream(httpx2.AsyncByteStream):
async def __aiter__(self) -> AsyncIterator[bytes]:
yield FIRST_CHUNK
raise httpx2.ReadTimeout("timed out while reading the stream")


def test_sync_stream_wraps_mid_stream_transport_error() -> None:
def handler(request: httpx2.Request) -> httpx2.Response:
return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=_DiesMidStream())

client = OpenAI(
api_key="My API Key",
http_client=httpx2.Client(transport=httpx2.MockTransport(handler)),
max_retries=0,
)

with pytest.raises(APITimeoutError):
for _ in client.chat.completions.create(
model="gpt-5.2", messages=[{"role": "user", "content": "hi"}], stream=True
):
pass


async def test_async_stream_wraps_mid_stream_transport_error() -> None:
async def handler(request: httpx2.Request) -> httpx2.Response:
return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=_AsyncDiesMidStream())

client = AsyncOpenAI(
api_key="My API Key",
http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)),
max_retries=0,
)

with pytest.raises(APITimeoutError):
async for _ in await client.chat.completions.create(
model="gpt-5.2", messages=[{"role": "user", "content": "hi"}], stream=True
):
pass


@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
Expand Down