DBOS.get_events (#477)

Adds new methods that retrieve all the published events for a given
workflow.

```python
def get_all_events(workflow_id: str) -> Dict[str, Any]:
async def get_all_events_async(workflow_id: str) -> Dict[str, Any]:
```

Addresses #476

Author: kraftp

dbos/_dbos.py

@@ -16,6 +16,7 @@
     AsyncGenerator,
     Callable,
     Coroutine,
+    Dict,
     Generator,
     Generic,
     List,
@@ -974,6 +975,33 @@ async def get_event_async(
             lambda: DBOS.get_event(workflow_id, key, timeout_seconds)
         )
 
+    @classmethod
+    def get_all_events(cls, workflow_id: str) -> Dict[str, Any]:
+        """
+        Get all events currently present for a workflow ID.
+        Args:
+            workflow_id: The workflow ID for which to get events
+        Returns:
+            A dictionary mapping event keys to their deserialized values
+        """
+
+        def fn() -> Dict[str, Any]:
+            return _get_dbos_instance()._sys_db.get_all_events(workflow_id)
+
+        return _get_dbos_instance()._sys_db.call_function_as_step(fn, "DBOS.get_events")
+
+    @classmethod
+    async def get_all_events_async(cls, workflow_id: str) -> Dict[str, Any]:
+        """
+        Get all events currently present for a workflow ID.
+        Args:
+            workflow_id: The workflow ID for which to get events
+        Returns:
+            A dictionary mapping event keys to their deserialized values
+        """
+        await cls._configure_asyncio_thread_pool()
+        return await asyncio.to_thread(cls.get_all_events, workflow_id)
+
     @classmethod
     def _execute_workflow_id(cls, workflow_id: str) -> WorkflowHandle[Any]:
         """Execute a workflow by ID (for recovery)."""
@@ -1225,7 +1253,7 @@ def fn() -> List[StepInfo]:
     async def list_workflow_steps_async(cls, workflow_id: str) -> List[StepInfo]:
         await cls._configure_asyncio_thread_pool()
         return await asyncio.to_thread(cls.list_workflow_steps, workflow_id)
-    
+
     @classproperty
     def application_version(cls) -> str:
         return GlobalParams.app_version

dbos/_sys_db.py

@@ -1567,6 +1567,32 @@ def set_event(
             }
             self._record_operation_result_txn(output, conn=c)
 
+    def get_all_events(self, workflow_id: str) -> Dict[str, Any]:
+        """
+        Get all events currently present for a workflow ID.
+
+        Args:
+            workflow_id: The workflow UUID to get events for
+
+        Returns:
+            A dictionary mapping event keys to their deserialized values
+        """
+        with self.engine.begin() as c:
+            rows = c.execute(
+                sa.select(
+                    SystemSchema.workflow_events.c.key,
+                    SystemSchema.workflow_events.c.value,
+                ).where(SystemSchema.workflow_events.c.workflow_uuid == workflow_id)
+            ).fetchall()
+
+            events: Dict[str, Any] = {}
+            for row in rows:
+                key = row[0]
+                value = _serialization.deserialize(row[1])
+                events[key] = value
+
+            return events
+
     @db_retry()
     def get_event(
         self,

tests/test_async.py

@@ -606,3 +606,43 @@ async def run_workflow_task() -> str:
     # Verify the workflow completes despite the task cancellation
     handle: WorkflowHandleAsync[str] = await DBOS.retrieve_workflow_async(wfid)
     assert await handle.get_result() == "completed"
+
+
+@pytest.mark.asyncio
+async def test_get_events_async(dbos: DBOS) -> None:
+    """Test the async version of get_events function that retrieves all events for a workflow."""
+
+    @DBOS.workflow()
+    async def async_events_workflow() -> str:
+        # Set multiple events using async methods
+        await DBOS.set_event_async("event1", "value1")
+        await DBOS.set_event_async("event2", {"nested": "data", "count": 42})
+        await DBOS.set_event_async("event3", [1, 2, 3, 4, 5])
+        return "completed"
+
+    # Execute the workflow
+    handle = await DBOS.start_workflow_async(async_events_workflow)
+    result = await handle.get_result()
+    assert result == "completed"
+
+    # Get all events for the workflow using async method
+    events = await DBOS.get_all_events_async(handle.workflow_id)
+
+    # Verify all events are present with correct values
+    assert len(events) == 3
+    assert events["event1"] == "value1"
+    assert events["event2"] == {"nested": "data", "count": 42}
+    assert events["event3"] == [1, 2, 3, 4, 5]
+
+    # Test with a workflow that has no events
+    @DBOS.workflow()
+    async def no_events_workflow() -> str:
+        await DBOS.sleep_async(0.01)
+        return "no events"
+
+    handle2 = await DBOS.start_workflow_async(no_events_workflow)
+    await handle2.get_result()
+
+    # Should return empty dict for workflow with no events
+    events2 = await DBOS.get_all_events_async(handle2.workflow_id)
+    assert events2 == {}

tests/test_dbos.py

@@ -1890,3 +1890,52 @@ def recv_workflow() -> Any:
     steps = client.list_workflow_steps(handle.workflow_id)
     assert len(steps) == 3
     assert "setEvent" in steps[0]["function_name"]
+
+
+def test_get_events(dbos: DBOS) -> None:
+
+    @DBOS.workflow()
+    def events_workflow() -> str:
+        # Set multiple events
+        DBOS.set_event("event1", "value1")
+        DBOS.set_event("event2", {"nested": "data", "count": 42})
+        DBOS.set_event("event3", [1, 2, 3, 4, 5])
+        return "completed"
+
+    # Execute the workflow
+    handle = DBOS.start_workflow(events_workflow)
+    result = handle.get_result()
+    assert result == "completed"
+
+    # Get events, verify they are present with correct values
+    def get_events() -> None:
+        events = DBOS.get_all_events(handle.workflow_id)
+
+        assert len(events) == 3
+        assert events["event1"] == "value1"
+        assert events["event2"] == {"nested": "data", "count": 42}
+        assert events["event3"] == [1, 2, 3, 4, 5]
+
+    # Verify it works
+    get_events()
+
+    # Run it as a workflow, verify it still works
+    get_events_workflow = DBOS.workflow()(get_events)
+    wfid = str(uuid.uuid4())
+    with SetWorkflowID(wfid):
+        get_events_workflow()
+    steps = DBOS.list_workflow_steps(wfid)
+    assert len(steps) == 1
+    assert steps[0]["function_name"] == "DBOS.get_events"
+
+    # Test with a workflow that has no events
+    @DBOS.workflow()
+    def no_events_workflow() -> str:
+        return "no events"
+
+    handle2 = DBOS.start_workflow(no_events_workflow)
+    handle2.get_result()
+
+    # Should return empty dict for workflow with no events
+    events2 = DBOS.get_all_events(handle2.workflow_id)
+    assert events2 == {}