Add get_current_history_length() to workflow info (#73)

Author: cretz

temporalio/worker/workflow_instance.py

@@ -149,6 +149,7 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         self._info = det.info
         self._primary_task: Optional[asyncio.Task[None]] = None
         self._time = 0.0
+        self._current_history_length = 0
         # Handles which are ready to run on the next event loop iteration
         self._ready: Deque[asyncio.Handle] = collections.deque()
         self._conditions: List[Tuple[Callable[[], bool], asyncio.Future]] = []
@@ -231,6 +232,7 @@ def activate(
         )
         self._current_completion.successful.SetInParent()
         self._current_activation_error: Optional[Exception] = None
+        self._current_history_length = act.history_length
         self._time = act.timestamp.ToMicroseconds() / 1e6
         self._is_replaying = act.is_replaying
 
@@ -652,6 +654,9 @@ def workflow_continue_as_new(
         # TODO(cretz): Why can't MyPy infer the above never returns?
         raise RuntimeError("Unreachable")
 
+    def workflow_get_current_history_length(self) -> int:
+        return self._current_history_length
+
     def workflow_get_external_workflow_handle(
         self, id: str, *, run_id: Optional[str]
     ) -> temporalio.workflow.ExternalWorkflowHandle[Any]:

temporalio/workflow.py

@@ -293,6 +293,14 @@ def _logger_details(self) -> Mapping[str, Any]:
             "workflow_type": self.workflow_type,
         }
 
+    def get_current_history_length(self) -> int:
+        """Get the current number of events in history.
+
+        Returns:
+            Current number of events in history (up until the current task).
+        """
+        return _Runtime.current().workflow_get_current_history_length()
+
 
 @dataclass(frozen=True)
 class ParentInfo:
@@ -347,6 +355,10 @@ def workflow_continue_as_new(
     ) -> NoReturn:
         ...
 
+    @abstractmethod
+    def workflow_get_current_history_length(self) -> int:
+        ...
+
     @abstractmethod
     def workflow_get_external_workflow_handle(
         self, id: str, *, run_id: Optional[str]

tests/worker/test_workflow.py

@@ -110,7 +110,9 @@ class InfoWorkflow:
     @workflow.run
     async def run(self) -> Dict:
         # Convert to JSON and back so it'll stringify un-JSON-able pieces
-        return json.loads(json.dumps(dataclasses.asdict(workflow.info()), default=str))
+        ret = dataclasses.asdict(workflow.info())
+        ret["current_history_length"] = workflow.info().get_current_history_length()
+        return json.loads(json.dumps(ret, default=str))
 
 
 async def test_workflow_info(client: Client):
@@ -130,6 +132,7 @@ async def test_workflow_info(client: Client):
         )
         assert info["attempt"] == 1
         assert info["cron_schedule"] is None
+        assert info["current_history_length"] == 3
         assert info["execution_timeout"] is None
         assert info["namespace"] == client.namespace
         assert info["retry_policy"] == json.loads(