diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 78e2d20aa7..6df14ccb6d 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -63,6 +63,8 @@ def __stream__(self) -> Iterator[_T]: for sse in iterator: if sse.data.startswith("[DONE]"): break + if not sse.data: + continue # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data if sse.event and sse.event.startswith("thread."): @@ -173,6 +175,8 @@ async def __stream__(self) -> AsyncIterator[_T]: async for sse in iterator: if sse.data.startswith("[DONE]"): break + if not sse.data: + continue # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data if sse.event and sse.event.startswith("thread."): diff --git a/tests/test_streaming.py b/tests/test_streaming.py index ae6c0590f7..983bd7e5a4 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -58,6 +58,31 @@ def body() -> Iterator[bytes]: await assert_empty_iter(iterator) +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_stream_skips_empty_sse_events( + sync: bool, + client: OpenAI, + async_client: AsyncOpenAI, +) -> None: + def body() -> Iterator[bytes]: + yield b"id: control-only\n" + yield b"retry: 1000\n" + yield b"\n" + yield b'data: {"foo":true}\n' + yield b"\n" + + if sync: + stream = Stream(cast_to=object, client=client, response=httpx2.Response(200, content=body())) + assert list(stream) == [{"foo": True}] + else: + stream = AsyncStream( + cast_to=object, + client=async_client, + response=httpx2.Response(200, content=to_aiter(body())), + ) + assert [item async for item in stream] == [{"foo": True}] + + @pytest.mark.asyncio @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) async def test_multiple_events(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: