flip logic around

Author: dtmeadows

examples/image_stream.py

@@ -50,4 +50,4 @@ def main() -> None:
     try:
         main()
     except Exception as error:
-        print(f"Error generating image: {error}")
+        print(f"Error generating image: {error}")

src/openai/_streaming.py

@@ -59,9 +59,11 @@ def __stream__(self) -> Iterator[_T]:
             if sse.data.startswith("[DONE]"):
                 break
 
-            if sse.event is not None and not sse.event.startswith("thread."):
+            # 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."):
                 data = sse.json()
-                if is_mapping(data) and data.get("error"):
+
+                if sse.event == "error" and is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -75,12 +77,10 @@ def __stream__(self) -> Iterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data=data, cast_to=cast_to, response=response)
-
+                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
             else:
                 data = sse.json()
-
-                if sse.event == "error" and is_mapping(data) and data.get("error"):
+                if is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -93,8 +93,8 @@ def __stream__(self) -> Iterator[_T]:
                         request=self.response.request,
                         body=data["error"],
                     )
-                # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data
-                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
+
+                yield process_data(data=data, cast_to=cast_to, response=response)
 
         # Ensure the entire stream is consumed
         for _sse in iterator:
@@ -161,9 +161,11 @@ async def __stream__(self) -> AsyncIterator[_T]:
             if sse.data.startswith("[DONE]"):
                 break
 
-            if sse.event is None or sse.event.startswith("response.") or sse.event.startswith("transcript."):
+            # 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."):
                 data = sse.json()
-                if is_mapping(data) and data.get("error"):
+
+                if sse.event == "error" and is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -177,12 +179,10 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data=data, cast_to=cast_to, response=response)
-
+                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
             else:
                 data = sse.json()
-
-                if sse.event == "error" and is_mapping(data) and data.get("error"):
+                if is_mapping(data) and data.get("error"):
                     message = None
                     error = data.get("error")
                     if is_mapping(error):
@@ -196,7 +196,7 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         body=data["error"],
                     )
 
-                yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response)
+                yield process_data(data=data, cast_to=cast_to, response=response)
 
         # Ensure the entire stream is consumed
         async for _sse in iterator: