Reference impl from client instead of interface

Author: dandavison

hello_nexus/caller/app.py

@@ -5,14 +5,6 @@
 from temporalio.client import Client
 from temporalio.worker import UnsandboxedWorkflowRunner, Worker
 
-from hello_nexus.caller.workflows_via_interface import (
-    Echo2CallerWorkflow,
-    Echo3CallerWorkflow,
-    EchoCallerWorkflow,
-    Hello2CallerWorkflow,
-    HelloCallerWorkflow,
-)
-
 interrupt_event = asyncio.Event()
 
 
@@ -39,11 +31,31 @@ async def execute_workflow(workflow_cls: Type[Any], input: Any) -> None:
 
 
 if __name__ == "__main__":
-    if len(sys.argv) != 2:
-        print("Usage: python -m nexus.caller.app [echo|hello]")
+    if len(sys.argv) != 3:
+        print("Usage: python -m nexus.caller.app [echo|hello] [impl|interface]")
         sys.exit(1)
 
-    [wf_name] = sys.argv[1:]
+    [wf_name, impl_or_interface] = sys.argv[1:]
+
+    if impl_or_interface == "impl":
+        from hello_nexus.caller.workflows_via_impl import (
+            Echo2CallerWorkflow,
+            Echo3CallerWorkflow,
+            EchoCallerWorkflow,
+            Hello2CallerWorkflow,
+            HelloCallerWorkflow,
+        )
+    elif impl_or_interface == "interface":
+        from hello_nexus.caller.workflows_via_interface import (
+            Echo2CallerWorkflow,
+            Echo3CallerWorkflow,
+            EchoCallerWorkflow,
+            Hello2CallerWorkflow,
+            HelloCallerWorkflow,
+        )
+    else:
+        raise ValueError(f"Invalid impl_or_interface: {impl_or_interface}")
+
     fn = {
         "echo": lambda: execute_workflow(EchoCallerWorkflow, "hello"),
         "echo2": lambda: execute_workflow(Echo2CallerWorkflow, "hello"),

hello_nexus/caller/workflows_via_impl.py

@@ -0,0 +1,93 @@
+from datetime import timedelta
+
+import xray
+from temporalio import workflow
+from temporalio.exceptions import FailureError
+from temporalio.workflow import NexusClient
+
+from hello_nexus.handler.nexus_service import (
+    EchoInput,
+    EchoOutput,
+    HelloInput,
+    HelloOutput,
+    MyNexusService,
+)
+
+
+class CallerWorkflowBase:
+    def __init__(self):
+        self.nexus_client = NexusClient(
+            MyNexusService,  # or string name "my-nexus-service",
+            "my-nexus-endpoint-name-python",
+            schedule_to_close_timeout=timedelta(seconds=30),
+        )
+
+
+@workflow.defn
+class EchoCallerWorkflow(CallerWorkflowBase):
+    @xray.start_as_current_workflow_method_span()
+    @workflow.run
+    async def run(self, message: str) -> EchoOutput:
+        op_output = await self.nexus_client.execute_operation(
+            MyNexusService.echo,
+            EchoInput(message),
+        )
+        return op_output
+
+
+@workflow.defn
+class Echo2CallerWorkflow(CallerWorkflowBase):
+    @xray.start_as_current_workflow_method_span()
+    @workflow.run
+    async def run(self, message: str) -> EchoOutput:
+        op_output = await self.nexus_client.execute_operation(
+            MyNexusService.echo2,
+            EchoInput(message),
+        )
+        return op_output
+
+
+@workflow.defn
+class Echo3CallerWorkflow(CallerWorkflowBase):
+    @xray.start_as_current_workflow_method_span()
+    @workflow.run
+    async def run(self, message: str) -> EchoOutput:
+        op_output = await self.nexus_client.execute_operation(
+            MyNexusService.echo3,
+            EchoInput(message),
+        )
+        return op_output
+
+
+@workflow.defn
+class HelloCallerWorkflow(CallerWorkflowBase):
+    @xray.start_as_current_workflow_method_span()
+    @workflow.run
+    async def run(self, name: str) -> HelloOutput:
+        handle = await self.nexus_client.start_operation(
+            MyNexusService.hello,
+            HelloInput(name),
+        )
+        assert handle.cancel()
+        try:
+            await handle
+        except FailureError:
+            handle = await self.nexus_client.start_operation(
+                MyNexusService.hello,
+                HelloInput(name),
+            )
+            result = await handle
+            return result
+        raise AssertionError("Expected Nexus operation to be cancelled")
+
+
+@workflow.defn
+class Hello2CallerWorkflow(CallerWorkflowBase):
+    @xray.start_as_current_workflow_method_span()
+    @workflow.run
+    async def run(self, name: str) -> HelloOutput:
+        handle = await self.nexus_client.start_operation(
+            MyNexusService.hello2,
+            HelloInput(name),
+        )
+        return await handle