Add WorkflowClient.new_workflow_stub_from_workflow_id()

Author: firdaus

README.md

@@ -28,7 +28,7 @@ version is targeted to be released in ~~September of 2019~~ ~~January 2020~~ Mar
 - [x] newRandom
 - [x] UUID
 - [x] Workflow Versioning
-- [ ] WorkflowClient.newWorkflowStub(Class workflowInterface, String workflowId);
+- [x] WorkflowClient.newWorkflowStub(Class workflowInterface, String workflowId);
 
 1.1
 - [ ] ActivityStub and Workflow.newUntypedActivityStub

cadence/tests/test_stub_workflow_id.py

@@ -0,0 +1,75 @@
+import time
+
+import pytest
+
+from cadence.exceptions import QueryFailureException
+from cadence.workerfactory import WorkerFactory
+from cadence.workflow import workflow_method, signal_method, Workflow, WorkflowClient, query_method
+
+TASK_LIST = "TestStubWorkflowId"
+DOMAIN = "sample"
+
+
+class GreetingException(Exception):
+    pass
+
+
+class TestStubWorkflowId:
+
+    @query_method()
+    async def get_message(self) -> str:
+        raise NotImplementedError
+
+    @query_method()
+    async def get_message_fail(self) -> str:
+        raise NotImplementedError
+
+    @signal_method()
+    async def put_message(self, message):
+        raise NotImplementedError
+
+    @workflow_method(task_list=TASK_LIST)
+    async def get_greetings(self) -> list:
+        raise NotImplementedError
+
+
+class TestStubWorkflowIdImpl(TestStubWorkflowId):
+
+    def __init__(self):
+        self.message = ""
+
+    async def get_message(self) -> str:
+        return self.message
+
+    async def get_message_fail(self) -> str:
+        raise GreetingException("error from query")
+
+    async def put_message(self, message):
+        self.message = message
+
+    async def get_greetings(self) -> list:
+        self.message = "initial-message"
+        await Workflow.await_till(lambda: self.message == "done")
+        return "finished"
+
+
+def test_stub_workflow_id():
+    factory = WorkerFactory("localhost", 7933, DOMAIN)
+    worker = factory.new_worker(TASK_LIST)
+    worker.register_workflow_implementation_type(TestStubWorkflowIdImpl)
+    factory.start()
+
+    client = WorkflowClient.new_client(domain=DOMAIN)
+    workflow: TestStubWorkflowId = client.new_workflow_stub(TestStubWorkflowId)
+    context = WorkflowClient.start(workflow.get_greetings)
+
+    stub: TestStubWorkflowId = client.new_workflow_stub_from_workflow_id(TestStubWorkflowId,
+                                                                         workflow_id=context.workflow_execution.workflow_id)
+    stub.put_message("abc")
+    assert stub.get_message() == "abc"
+
+    stub.put_message("done")
+    assert client.wait_for_close_with_workflow_id(context.workflow_execution.workflow_id) == "finished"
+
+    print("Stopping workers")
+    worker.stop()

cadence/workflow.py

@@ -133,10 +133,24 @@ def new_workflow_stub(self, cls: Type, workflow_options: WorkflowOptions = None)
         stub_cls = type(cls.__name__, (WorkflowStub,), attrs)
         return stub_cls()
 
+    def new_workflow_stub_from_workflow_id(self, cls: Type, workflow_id: str):
+        """
+        Use it to send signals or queries to a running workflow.
+        Do not call workflow methods on it
+        """
+        stub_instance = self.new_workflow_stub(cls)
+        execution = WorkflowExecution(workflow_id=workflow_id, run_id=None)
+        stub_instance._execution = execution
+        return stub_instance
+
     def wait_for_close(self, context: WorkflowExecutionContext) -> object:
+        return self.wait_for_close_with_workflow_id(workflow_id=context.workflow_execution.workflow_id,
+                                                    run_id=context.workflow_execution.run_id,
+                                                    workflow_type=context.workflow_type)
+
+    def wait_for_close_with_workflow_id(self, workflow_id: str, run_id: str = None, workflow_type: str = None):
         while True:
-            history_request = create_close_history_event_request(self, context.workflow_execution.workflow_id,
-                                                                 context.workflow_execution.run_id)
+            history_request = create_close_history_event_request(self, workflow_id, run_id)
             history_response, err = self.service.get_workflow_execution_history(history_request)
             if err:
                 raise Exception(err)
@@ -152,8 +166,9 @@ def wait_for_close(self, context: WorkflowExecutionContext) -> object:
                     exception = deserialize_exception(attributes.details)
                     if isinstance(exception, ActivityFailureException):
                         exception.set_cause()
-                    raise WorkflowFailureException(workflow_type=context.workflow_type,
-                                                   execution=context.workflow_execution) from exception
+                    workflow_execution = WorkflowExecution(workflow_id=workflow_id, run_id=run_id)
+                    raise WorkflowFailureException(workflow_type=workflow_type,
+                                                   execution=workflow_execution) from exception
                 else:
                     details: Dict = json.loads(attributes.details)
                     detail_message = details.get("detailMessage", "")