fix: sync iter (#225)

* fix: sync iter

* release: 0.38.3

* fix: ensure we have a loop

Author: grutt

examples/fanout/sync_stream.py

@@ -0,0 +1,47 @@
+import asyncio
+import base64
+import json
+import os
+import random
+
+from dotenv import load_dotenv
+
+from hatchet_sdk import new_client
+from hatchet_sdk.clients.admin import TriggerWorkflowOptions
+from hatchet_sdk.clients.run_event_listener import StepRunEventType
+from hatchet_sdk.v2.hatchet import Hatchet
+
+
+def main():
+    load_dotenv()
+    hatchet = Hatchet()
+
+    # Generate a random stream key to use to track all
+    # stream events for this workflow run.
+
+    streamKey = "streamKey"
+    streamVal = f"sk-{random.randint(1, 100)}"
+
+    # Specify the stream key as additional metadata
+    # when running the workflow.
+
+    # This key gets propagated to all child workflows
+    # and can have an arbitrary property name.
+
+    workflowRun = hatchet.admin.run_workflow(
+        "Parent",
+        {"n": 2},
+        options={"additional_metadata": {streamKey: streamVal}},
+    )
+
+    # Stream all events for the additional meta key value
+    listener = hatchet.listener.stream_by_additional_metadata(streamKey, streamVal)
+
+    for event in listener:
+        print(event.type, event.payload)
+
+    print("DONE.")
+
+
+if __name__ == "__main__":
+    main()

hatchet_sdk/clients/run_event_listener.py

@@ -96,6 +96,31 @@ def abort(self):
     def __aiter__(self):
         return self._generator()
 
+    async def __anext__(self):
+        return await self._generator().__anext__()
+
+    def __iter__(self):
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError as e:
+            if str(e).startswith("There is no current event loop in thread"):
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+            else:
+                raise e
+
+        async_iter = self.__aiter__()
+
+        while True:
+            try:
+                future = asyncio.ensure_future(async_iter.__anext__())
+                yield loop.run_until_complete(future)
+            except StopAsyncIteration:
+                break
+            except Exception as e:
+                print(f"Error in synchronous iterator: {e}")
+                break
+
     async def _generator(self) -> AsyncGenerator[StepRunEvent, None]:
         while True:
             if self.stop_signal:
@@ -167,6 +192,7 @@ async def _generator(self) -> AsyncGenerator[StepRunEvent, None]:
                     # Unknown error, report and break
                     # logger.error(f"Failed to receive message: {e}")
                     break
+                # Raise StopAsyncIteration to properly end the generator
 
     async def retry_subscribe(self):
         retries = 0

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hatchet-sdk"
-version = "0.38.2"
+version = "0.38.3"
 description = ""
 authors = ["Alexander Belanger <[email protected]>"]
 readme = "README.md"