Implement Workflow.new_untyped_activity_stub().

Author: firdaus

temporal/activity_method.py

@@ -94,3 +94,21 @@ def fill_execute_activity_parameters(self, execute_parameters: ExecuteActivityPa
             execute_parameters.heartbeat_timeout = self.heartbeat_timeout
         if self.task_queue is not None:
             execute_parameters.task_queue = self.task_queue
+
+
+@dataclass
+class UntypedActivityStub:
+    _decision_context: object = None
+    _retry_parameters: RetryParameters = None
+    _activity_options: ActivityOptions = None
+
+    async def execute(self, activity_name: str, *args):
+        f = await self.execute_async(activity_name, *args)
+        return await f.wait_for_result()
+
+    async def execute_async(self, activity_name: str, *args):
+        from .async_activity import Async
+        execute_parameters = ExecuteActivityParameters()
+        execute_parameters.activity_type = ActivityType()
+        execute_parameters.activity_type.name = activity_name
+        return Async.call(self, execute_parameters, args)

temporal/async_activity.py

@@ -17,6 +17,10 @@ def function_with_self(method, self, *args):
         assert self._decision_context
         assert method._execute_parameters
         parameters: ExecuteActivityParameters = copy.deepcopy(method._execute_parameters)
+        return Async.call(self, parameters, args)
+
+    @staticmethod
+    def call(self, parameters, args: List[object]):
         if hasattr(self, "_activity_options") and self._activity_options:
             self._activity_options.fill_execute_activity_parameters(parameters)
         if self._retry_parameters:

temporal/workflow.py

@@ -41,6 +41,19 @@ def new_activity_stub(activities_cls, retry_parameters: RetryParameters = None,
         cls._activity_options = activity_options
         return cls
 
+    @staticmethod
+    def new_untyped_activity_stub(retry_parameters: RetryParameters = None,
+                                  activity_options: ActivityOptions = None):
+        from .decision_loop import ITask
+        from .activity_method import UntypedActivityStub
+        task: ITask = ITask.current()
+        assert task
+        cls = UntypedActivityStub()
+        cls._decision_context = task.decider.decision_context
+        cls._retry_parameters = retry_parameters  # type: ignore
+        cls._activity_options = activity_options
+        return cls
+
     @staticmethod
     async def await_till(c: Callable, timeout_seconds: int = 0) -> bool:
         from .decision_loop import ITask

tests/test_workflow_untyped_activity_stub.py

@@ -0,0 +1,49 @@
+import pytest
+from datetime import timedelta
+
+from temporal.workflow import workflow_method, WorkflowClient, Workflow
+from temporal.activity_method import activity_method, ActivityOptions
+
+TASK_QUEUE = "test_workflow_untyped_activity_stub_tq"
+NAMESPACE = "default"
+
+
+class GreetingActivities:
+    @activity_method(task_queue=TASK_QUEUE, schedule_to_close_timeout=timedelta(seconds=1000))
+    async def compose_greeting(self, arg1) -> str:
+        raise NotImplementedError
+
+
+class GreetingActivitiesImpl:
+
+    def compose_greeting(self, arg1):
+        return "from-activity: " + arg1
+
+
+class GreetingWorkflow:
+    @workflow_method(task_queue=TASK_QUEUE)
+    async def get_greeting(self) -> None:
+        raise NotImplementedError
+
+
+class GreetingWorkflowImpl(GreetingWorkflow):
+
+    def __init__(self):
+        self.stub = Workflow.new_untyped_activity_stub(
+            activity_options=ActivityOptions(task_queue=TASK_QUEUE,
+                                             schedule_to_close_timeout=timedelta(seconds=1000))
+        )
+
+    async def get_greeting(self):
+        return await self.stub.execute("GreetingActivities::compose_greeting", "blah")
+
+
+@pytest.mark.asyncio
+@pytest.mark.worker_config(NAMESPACE, TASK_QUEUE, activities=[(GreetingActivitiesImpl(), "GreetingActivities")],
+                           workflows=[GreetingWorkflowImpl])
+async def test(worker):
+    client = WorkflowClient.new_client(namespace=NAMESPACE)
+    greeting_workflow: GreetingWorkflow = client.new_workflow_stub(GreetingWorkflow)
+    ret_value = await greeting_workflow.get_greeting()
+
+    assert ret_value == "from-activity: blah"