Skip empty SSE control events

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."):

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: