Nexus samples

Author: dandavison

hello_nexus/README.md

@@ -0,0 +1,15 @@
+# Nexus
+
+Temporal Nexus is a 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. These can abstract underlying Temporal primitives such as
+Workflows, or execute arbitrary code.
+
+Learn more at [temporal.io/nexus](https://temporal.io/nexus).
+
+The samples in this directory form an introduction to Nexus. 
+
+### Samples
+
+- [basic](./basic) - Nexus service definition, operation handlers, and calling workflows.
+- [without_service_definition](./without_service_definition) - A Nexus service implementation without a service definition 

hello_nexus/__init__.py

@@ -0,0 +1,37 @@
+This sample shows how to define a Nexus service, implement the operation handlers, and
+call the operations from a workflow.
+
+### Sample directory structure
+
+- [service.py](./service.py) - shared Nexus service definition
+- [caller](./caller) - a caller workflow that executes Nexus operations, together with a worker and starter code
+- [handler](./handler) - Nexus operation handlers, together with a workflow used by one of the Nexus operations, and a worker that polls for both workflow and Nexus tasks.
+
+
+### Instructions
+
+Start a Temporal server.
+
+Run the following:
+
+```
+temporal operator namespace create --namespace my-target-namespace
+temporal operator namespace create --namespace my-caller-namespace
+
+temporal operator nexus endpoint create \
+  --name my-nexus-endpoint \
+  --target-namespace my-target-namespace \
+  --target-task-queue my-target-task-queue \
+  --description-file ./hello_nexus/basic/service_description.md
+```
+
+In one terminal, run the Temporal worker in the handler namespace:
+```
+uv run hello_nexus/basic/handler/worker.py
+```
+
+In another terminal, run the Temporal worker in the caller namespace and start the caller
+workflow:
+```
+uv run hello_nexus/basic/caller/app.py
+```

hello_nexus/basic/README.md

@@ -0,0 +1 @@
+from . import app as app

hello_nexus/basic/__init__.py

@@ -0,0 +1,46 @@
+import asyncio
+import uuid
+from typing import Optional
+
+from temporalio.client import Client
+from temporalio.worker import UnsandboxedWorkflowRunner, Worker
+
+from hello_nexus.basic.caller.workflows import CallerWorkflow
+from hello_nexus.basic.service import MyOutput
+
+NAMESPACE = "my-caller-namespace"
+TASK_QUEUE = "my-caller-task-queue"
+
+
+async def execute_caller_workflow(
+    client: Optional[Client] = None,
+) -> tuple[MyOutput, MyOutput]:
+    client = client or await Client.connect(
+        "localhost:7233",
+        namespace=NAMESPACE,
+    )
+
+    async with Worker(
+        client,
+        task_queue=TASK_QUEUE,
+        workflows=[CallerWorkflow],
+        # TODO(dan): isinstance(op, nexusrpc.contract.Operation) is failing under the
+        # sandbox in temporalio/worker/_interceptor.py
+        workflow_runner=UnsandboxedWorkflowRunner(),
+    ):
+        return await client.execute_workflow(
+            CallerWorkflow.run,
+            arg="world",
+            id=str(uuid.uuid4()),
+            task_queue=TASK_QUEUE,
+        )
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        results = loop.run_until_complete(execute_caller_workflow())
+        for output in results:
+            print(output.message)
+    except KeyboardInterrupt:
+        loop.run_until_complete(loop.shutdown_asyncgens())

hello_nexus/basic/caller/__init__.py

@@ -0,0 +1,27 @@
+from temporalio import workflow
+from temporalio.workflow import NexusServiceClient
+
+from hello_nexus.basic.service import MyInput, MyNexusService, MyOutput
+
+NEXUS_ENDPOINT = "my-nexus-endpoint"
+
+
+@workflow.defn
+class CallerWorkflow:
+    def __init__(self):
+        self.nexus_service_client = NexusServiceClient(
+            MyNexusService,
+            endpoint=NEXUS_ENDPOINT,
+        )
+
+    @workflow.run
+    async def run(self, name: str) -> tuple[MyOutput, MyOutput]:
+        sync_result: MyOutput = await self.nexus_service_client.execute_operation(
+            MyNexusService.my_sync_operation,
+            MyInput(name),
+        )
+        wf_result: MyOutput = await self.nexus_service_client.execute_operation(
+            MyNexusService.my_workflow_run_operation,
+            MyInput(name),
+        )
+        return sync_result, wf_result

hello_nexus/basic/caller/app.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+
+class MyDBClient:
+    """
+    This class represents a resource that your Nexus operation handlers may need when they
+    are handling Nexus requests, but which is only available when the Nexus worker is
+    started. Notice that:
+
+    (a) The user's service handler class __init__ constructor takes a MyDBClient instance
+        (see hello_nexus.handler.MyNexusService)
+
+    (b) The user is responsible for instantiating the service handler class when they
+        start the worker (see hello_nexus.handler.worker), so they can pass any
+        necessary resources (such as this database client) to the service handler.
+    """
+
+    @classmethod
+    def connect(cls) -> MyDBClient:
+        return cls()
+
+    def execute(self, query: str) -> str:
+        return "query-result"

hello_nexus/basic/caller/workflows.py

@@ -0,0 +1,63 @@
+"""
+This file demonstrates how to define operation handlers by implementing the `start`
+method only, using the "shorthand" decorators sync_operation_handler and
+workflow_run_operation_handler.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import temporalio.common
+import temporalio.nexus.handler
+from nexusrpc.handler import (
+    StartOperationContext,
+    service_handler,
+    sync_operation_handler,
+)
+from temporalio.client import WorkflowHandle
+
+from hello_nexus.basic.handler.db_client import MyDBClient
+from hello_nexus.basic.handler.workflows import WorkflowStartedByNexusOperation
+from hello_nexus.basic.service import MyInput, MyNexusService, MyOutput
+
+
+@service_handler(service=MyNexusService)
+class MyNexusServiceHandler:
+    # You can create an __init__ method accepting what is needed by your operation
+    # handlers to handle requests. You typically instantiate your service handler class
+    # when starting your worker. See hello_nexus/basic/handler/worker.py.
+    def __init__(self, connected_db_client: MyDBClient):
+        # `connected_db_client` is intended as an example of something that might be
+        # required by your operation handlers when handling requests, but is only
+        # available at worker-start time.
+        self.connected_db_client = connected_db_client
+
+    # This is a nexus operation that is backed by a Temporal workflow. The start method
+    # starts a workflow, and returns a nexus operation token that the handler can use to
+    # obtain a workflow handle (for example if a cancel request is subsequently sent by
+    # the caller). The Temporal server takes care of delivering the workflow result to the
+    # calling workflow. The task queue defaults to the task queue being used by the Nexus
+    # worker.
+    @temporalio.nexus.handler.workflow_run_operation_handler
+    async def my_workflow_run_operation(
+        self, ctx: StartOperationContext, input: MyInput
+    ) -> WorkflowHandle[WorkflowStartedByNexusOperation, MyOutput]:
+        # You could use self.connected_db_client here.
+        return await temporalio.nexus.handler.start_workflow(
+            ctx,
+            WorkflowStartedByNexusOperation.run,
+            input,
+            id=str(uuid.uuid4()),
+        )
+
+    # This is a sync operation. That means that unlike the workflow run operation above,
+    # in this case the `start` method returns the final operation result. Sync operations
+    # are free to make arbitrary network calls, or perform CPU-bound computations. Total
+    # execution duration must not exceed 10s.
+    @sync_operation_handler
+    async def my_sync_operation(
+        self, ctx: StartOperationContext, input: MyInput
+    ) -> MyOutput:
+        # You could use self.connected_db_client here.
+        return MyOutput(message=f"Hello {input.name} from sync operation!")