Nexus prototype

uv add --editable ../sdk-python
uv add --group nexus --editable ../nexus-sdk-python

Author: dandavison

Diff for: hello_nexus/README.md

@@ -0,0 +1,70 @@
+# nexus
+
+Temporal Nexus is a new feature of the Temporal platform designed to connect durable executions across team, namespace,
+region, and cloud boundaries. It promotes a more modular architecture for sharing a subset of your team’s capabilities
+via well-defined service API contracts for other teams to use, that abstract underlying Temporal primitives, like
+Workflows, or execute arbitrary code.
+
+Learn more at [temporal.io/nexus](https://temporal.io/nexus).
+
+This sample shows how to use Temporal for authoring a Nexus service and call it from a workflow.
+
+### Sample directory structure
+
+- [service](./service) - shared service defintion
+- [caller](./caller) - caller workflows, worker, and starter
+- [handler](./handler) - handler workflow, operations, and worker
+- [options](./options) - command line argument parsing utility
+
+## Getting started locally
+
+### Get `temporal` CLI to enable local development
+
+1. Follow the instructions on the [docs
+   site](https://learn.temporal.io/getting_started/go/dev_environment/#set-up-a-local-temporal-service-for-development-with-temporal-cli)
+   to install Temporal CLI.
+
+> NOTE: Required version is at least v1.1.0.
+
+### Spin up environment
+
+#### Start temporal server
+
+> HTTP port is required for Nexus communications
+
+```
+temporal server start-dev --http-port 7243 --dynamic-config-value system.enableNexus=true
+```
+
+### Initialize environment
+
+#### Create caller and target namespaces
+
+```
+temporal operator namespace create --namespace my-target-namespace
+temporal operator namespace create --namespace my-caller-namespace
+```
+
+#### Create Nexus endpoint
+
+```
+temporal operator nexus endpoint create \
+  --name my-nexus-endpoint-name \
+  --target-namespace my-target-namespace \
+  --target-task-queue my-target-task-queue \
+  --description-file ./nexus/service/description.md
+```
+
+### Start Nexus handler worker
+```
+uv run python nexus/handler/worker.py
+```
+
+### Run Nexus caller application (worker + starter)
+```
+uv run python nexus/caller/app.py
+```
+
+### Output
+
+TODO

Diff for: hello_nexus/__init__.py

@@ -0,0 +1,3 @@
+from rich.traceback import install
+
+install(show_locals=True)

Diff for: hello_nexus/caller/app.py

@@ -0,0 +1,58 @@
+import asyncio
+import sys
+from typing import Any, Type
+
+from temporalio.client import Client
+from temporalio.worker import UnsandboxedWorkflowRunner, Worker
+
+from hello_nexus.caller.workflows import (
+    Echo2CallerWorkflow,
+    Echo3CallerWorkflow,
+    EchoCallerWorkflow,
+    Hello2CallerWorkflow,
+    HelloCallerWorkflow,
+)
+
+interrupt_event = asyncio.Event()
+
+
+async def execute_workflow(workflow_cls: Type[Any], input: Any) -> None:
+    client = await Client.connect("localhost:7233", namespace="my-caller-namespace")
+    task_queue = "my-caller-task-queue"
+
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[workflow_cls],
+        workflow_runner=UnsandboxedWorkflowRunner(),
+    ):
+        print("🟠 Caller worker started")
+        result = await client.execute_workflow(
+            workflow_cls.run,
+            input,
+            id="my-caller-workflow-id",
+            task_queue=task_queue,
+        )
+        print("🟢 workflow result:", result)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        print("Usage: python -m nexus.caller.app [echo|hello]")
+        sys.exit(1)
+
+    [wf_name] = sys.argv[1:]
+    fn = {
+        "echo": lambda: execute_workflow(EchoCallerWorkflow, "hello"),
+        "echo2": lambda: execute_workflow(Echo2CallerWorkflow, "hello"),
+        "echo3": lambda: execute_workflow(Echo3CallerWorkflow, "hello"),
+        "hello": lambda: execute_workflow(HelloCallerWorkflow, "world"),
+        "hello2": lambda: execute_workflow(Hello2CallerWorkflow, "world"),
+    }[wf_name]
+
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(fn())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())

Diff for: hello_nexus/caller/workflows.py

@@ -0,0 +1,104 @@
+from datetime import timedelta
+
+from temporalio import workflow
+from temporalio.exceptions import FailureError
+from temporalio.workflow import NexusClient
+
+from hello_nexus.service.interface 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",
+            schedule_to_close_timeout=timedelta(seconds=30),
+        )
+
+
+@workflow.defn
+class EchoCallerWorkflow(CallerWorkflowBase):
+    @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):
+    @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):
+    @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):
+    @workflow.run
+    async def run(self, name: str) -> HelloOutput:
+        # TODO: Java returns a handle immediately. The handle has a blocking method to
+        # wait until the operation has started (i.e. initial Nexus RPC response is
+        # available, so operation ID is available in the case of an async operation).
+        handle = await self.nexus_client.start_operation(
+            MyNexusService.hello,
+            HelloInput(name),
+        )
+        assert handle.cancel()
+        try:
+            await handle
+        # TODO(dan): But should this be asyncio.CancelledError? (there's also
+        # temporalio.exceptions.CancelledError which doesn't inherit from
+        # asyncio.CancelledError; as a workflow author, this confused me)
+        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):
+    @workflow.run
+    async def run(self, name: str) -> HelloOutput:
+        handle = await self.nexus_client.start_operation(
+            MyNexusService.hello2,
+            HelloInput(name),
+        )
+        return await handle
+
+
+@workflow.defn
+class Hello3CallerWorkflow(CallerWorkflowBase):
+    @workflow.run
+    async def run(self, name: str) -> HelloOutput:
+        handle = await self.nexus_client.start_operation(
+            MyNexusService.hello3,
+            HelloInput(name),
+        )
+        return await handle

Diff for: hello_nexus/clean

@@ -0,0 +1,14 @@
+temporal-delete-all my-target-namespace
+temporal-delete-all my-caller-namespace
+
+temporal operator namespace create --namespace my-target-namespace
+temporal operator namespace create --namespace my-caller-namespace
+
+sleep 1
+
+temporal operator nexus endpoint create \
+  --name my-nexus-endpoint-name \
+  --target-namespace my-target-namespace \
+  --target-task-queue my-target-task-queue \
+  --description-file ./hello_nexus/service/description.md
+  

Diff for: hello_nexus/handler/activities.py

@@ -0,0 +1,14 @@
+import asyncio
+
+from temporalio import activity
+
+from hello_nexus.service.interface import (
+    HelloInput,
+    HelloOutput,
+)
+
+
+@activity.defn
+async def hello_activity(input: HelloInput) -> HelloOutput:
+    await asyncio.sleep(1)
+    return HelloOutput(message=f"hello {input.name}")

Diff for: hello_nexus/handler/dbclient.py

@@ -0,0 +1,3 @@
+class MyDBClient:
+    def execute(self, query: str) -> str:
+        return "<query result>"