IWF-468: Describe Workflow API

Author: samuel27m

iwf/client.py

@@ -18,6 +18,7 @@
 from iwf.utils.iwf_typing import unset_to_none
 from iwf.utils.persistence_utils import get_search_attribute_value
 from iwf.workflow import ObjectWorkflow, get_workflow_type_by_class
+from iwf.workflow_info import WorkflowInfo
 from iwf.workflow_options import WorkflowOptions
 from iwf.workflow_state import (
     WorkflowState,
@@ -199,6 +200,19 @@ def reset_workflow(
             workflow_id, "", reset_workflow_type_and_options
         )
 
+    def describe_workflow(
+        self,
+        workflow_id: str,
+        workflow_run_id: Optional[str] = None,
+    ):
+        run_id = workflow_run_id if workflow_run_id is not None else ""
+
+        response = self._unregistered_client.get_workflow(
+            workflow_id, run_id
+        )
+
+        return WorkflowInfo(workflow_status=response.workflow_status)
+
     def skip_timer_by_command_id(
         self,
         workflow_id: str,

iwf/tests/test_describe_workflow.py

@@ -0,0 +1,68 @@
+import inspect
+import time
+import unittest
+
+from iwf.client import Client
+from iwf.command_request import CommandRequest, TimerCommand
+from iwf.command_results import CommandResults
+from iwf.communication import Communication
+from iwf.errors import WorkflowNotExistsError
+from iwf.iwf_api.models import WorkflowStatus
+from iwf.persistence import Persistence
+from iwf.state_decision import StateDecision
+from iwf.state_schema import StateSchema
+from iwf.tests.worker_server import registry
+from iwf.workflow import ObjectWorkflow
+from iwf.workflow_context import WorkflowContext
+from iwf.workflow_state import T, WorkflowState
+
+
+class WaitState(WorkflowState[None]):
+    def wait_until(
+        self,
+        ctx: WorkflowContext,
+        input: T,
+        persistence: Persistence,
+        communication: Communication,
+    ) -> CommandRequest:
+        return CommandRequest.for_all_command_completed(
+            TimerCommand.by_seconds(10),
+        )
+
+    def execute(
+        self,
+        ctx: WorkflowContext,
+        input: T,
+        command_results: CommandResults,
+        persistence: Persistence,
+        communication: Communication,
+    ) -> StateDecision:
+
+        return StateDecision.graceful_complete_workflow()
+
+
+class DescribeWorkflow(ObjectWorkflow):
+    def get_workflow_states(self) -> StateSchema:
+        return StateSchema.with_starting_state(WaitState())
+
+
+wf = DescribeWorkflow()
+registry.add_workflow(wf)
+client = Client(registry)
+
+class TestDescribeWorkflow(unittest.TestCase):
+    def test_describe_workflow(self):
+        wf_id = f"{inspect.currentframe().f_code.co_name}-{time.time_ns()}"
+
+        client.start_workflow(DescribeWorkflow, wf_id, 100)
+        workflow_info = client.describe_workflow(wf_id)
+        assert workflow_info.workflow_status == WorkflowStatus.RUNNING
+
+        # Stop the workflow
+        client.stop_workflow(wf_id)
+
+    def test_describe_workflow_when_workflow_not_exists(self):
+        wf_id = f"{inspect.currentframe().f_code.co_name}-{time.time_ns()}"
+
+        with self.assertRaises(WorkflowNotExistsError):
+            client.describe_workflow(wf_id)

iwf/unregistered_client.py

@@ -17,6 +17,7 @@
 from iwf.iwf_api import Client, errors
 from iwf.iwf_api.api.default import (
     post_api_v1_workflow_dataobjects_get,
+    post_api_v1_workflow_get,
     post_api_v1_workflow_reset,
     post_api_v1_workflow_rpc,
     post_api_v1_workflow_search,
@@ -96,7 +97,7 @@ def add_http_status(name, value, phrase, description=""):
         HTTPStatus.__class__
     )  # store the enum's member parent class
     setattr(HTTPStatus, name, new_status)  # add it to the global HTTPStatus namespace
-    HTTPStatus._member_map_[name] = new_status  #  add it to the name=>member map
+    HTTPStatus._member_map_[name] = new_status  # add it to the name=>member map
     HTTPStatus._member_names_.append(
         name
     )  # append the names so it appears in __members__
@@ -415,6 +416,21 @@ def stop_workflow(
         )
         return handler_error_and_return(response)
 
+    def get_workflow(
+        self,
+        workflow_id: str,
+        workflow_run_id: str,
+    ) -> WorkflowGetResponse:
+        request = WorkflowGetRequest(
+            workflow_id=workflow_id,
+            workflow_run_id=workflow_run_id,
+        )
+        response = post_api_v1_workflow_get.sync_detailed(
+            client=self.api_client,
+            json_body=request,
+        )
+        return handler_error_and_return(response)
+
     def get_workflow_data_attributes(
         self,
         workflow_id: str,

iwf/workflow_info.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+from iwf.iwf_api.models.workflow_status import WorkflowStatus
+
+
+@dataclass
+class WorkflowInfo:
+    workflow_status: WorkflowStatus