Nexus samples

Author: dandavison

hello_nexus/basic/README.md

@@ -0,0 +1,34 @@
+Usually you will want to create a service definition to formalize the service contract.
+However it is possible to define a Nexus service and operation handlers without creating a
+service definition. This sample demonstrates how to do that. This may be appropriate if
+you want to call a Nexus operation that is being executed by a Worker in the same
+namespace as the caller: in other words, if the Nexus operation is playing a role similar
+to an Activity.
+
+### 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-name-python \
+  --target-namespace my-target-namespace \
+  --target-task-queue my-target-task-queue-python \
+  --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/caller/app.py

@@ -0,0 +1,39 @@
+import asyncio
+import uuid
+
+from temporalio.client import Client
+from temporalio.worker import UnsandboxedWorkflowRunner, Worker
+
+from hello_nexus.basic.caller.workflows import CallerWorkflow
+
+
+async def main() -> 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=[CallerWorkflow],
+        # TODO(dan): isinstance(op, nexusrpc.contract.Operation) is failing under the
+        # sandbox in temporalio/worker/_interceptor.py
+        workflow_runner=UnsandboxedWorkflowRunner(),
+    ):
+        result = await client.execute_workflow(
+            CallerWorkflow.run,
+            arg="world",
+            id=str(uuid.uuid4()),
+            task_queue=task_queue,
+        )
+        print("🟢 workflow result:", result)
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        loop.run_until_complete(loop.shutdown_asyncgens())

hello_nexus/basic/caller/workflows.py

@@ -0,0 +1,27 @@
+from temporalio import workflow
+from temporalio.workflow import NexusServiceClient
+
+from hello_nexus.basic.service import (
+    MyInput,
+    MyNexusService,
+)
+
+
+@workflow.defn
+class CallerWorkflow:
+    def __init__(self):
+        self.nexus_service_client = NexusServiceClient(
+            MyNexusService,
+            endpoint="my-nexus-endpoint-name-python",
+        )
+
+    @workflow.run
+    async def run(self, name: str) -> None:
+        await self.nexus_service_client.execute_operation(
+            MyNexusService.my_sync_operation,
+            MyInput(name),
+        )
+        await self.nexus_service_client.start_operation(
+            MyNexusService.my_workflow_run_operation,
+            MyInput(name),
+        )

hello_nexus/basic/handler/db_client.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/handler/service_handler.py

@@ -0,0 +1,66 @@
+"""
+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. The `start` method actually returns the result.
+    # Implementations are free to make arbitrary network calls, or perform CPU-bound
+    # computations such as this one. 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}!")

hello_nexus/basic/handler/service_handler_with_operation_handler_classes.py

@@ -0,0 +1,143 @@
+"""
+Notes:
+
+Sync operations:
+---------------
+Implementations are free to make arbitrary network calls, or perform CPU-bound
+computations such as this one. Total execution duration must not exceed 10s. To
+perform Temporal client calls such as signaling/querying/listing workflows, use
+self.client.
+
+
+Workflow operations:
+---------------------
+The task queue defaults to the task queue being used by the Nexus worker.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import temporalio.common
+import temporalio.nexus.handler
+from nexusrpc.handler import (
+    CancelOperationContext,
+    FetchOperationInfoContext,
+    FetchOperationResultContext,
+    OperationHandler,
+    OperationInfo,
+    StartOperationContext,
+    StartOperationResultAsync,
+    StartOperationResultSync,
+    operation_handler,
+    service_handler,
+)
+
+from hello_nexus.basic.handler.db_client import MyDBClient
+from hello_nexus.basic.handler.service_handler import (
+    MyInput,
+    MyNexusService,
+    MyOutput,
+)
+from hello_nexus.basic.handler.workflows import WorkflowStartedByNexusOperation
+
+
+@service_handler(service=MyNexusService)
+class MyNexusServiceHandlerUsingOperationHandlerClasses:
+    # 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
+
+    @operation_handler
+    def my_sync_operation(self) -> OperationHandler[MyInput, MyOutput]:
+        # Pass any required arguments to the OperationHandler __init__ method here.
+        return MySyncOperation()
+
+    @operation_handler
+    def my_workflow_run_operation(
+        self,
+    ) -> OperationHandler[MyInput, MyOutput]:
+        # Pass any required arguments to the OperationHandler __init__ method here.
+        return MyWorkflowRunOperation()
+
+
+class MySyncOperation(OperationHandler[MyInput, MyOutput]):
+    # You can add an __init__ method taking any required arguments, since you are in
+    # control of instantiating the OperationHandler inside the operation handler method
+    # above decorated with @operation_handler.
+
+    async def start(
+        self, ctx: StartOperationContext, input: MyInput
+    ) -> StartOperationResultSync[MyOutput]:
+        output = MyOutput(message=f"Hello {input.name}!")
+        return StartOperationResultSync(output)
+
+    async def fetch_info(
+        self,
+        ctx: FetchOperationInfoContext,
+        token: str,
+    ) -> OperationInfo:
+        raise NotImplementedError(
+            "fetch_info is not supported when a Nexus operation is called by a Temporal workflow"
+        )
+
+    async def fetch_result(
+        self,
+        ctx: FetchOperationResultContext,
+        token: str,
+    ) -> MyOutput:
+        raise NotImplementedError(
+            "fetch_result is not supported when a Nexus operation is called by a Temporal workflow, "
+            "but this sample does not demonstrate result fetching"
+        )
+
+    async def cancel(
+        self,
+        ctx: CancelOperationContext,
+        token: str,
+    ) -> None:
+        raise NotImplementedError(
+            "cancel is supported when a Nexus operation is called by a Temporal workflow, "
+            "but this sample does not demonstrate cancellation"
+        )
+
+
+class MyWorkflowRunOperation(OperationHandler[MyInput, MyOutput]):
+    async def start(
+        self, ctx: StartOperationContext, input: MyInput
+    ) -> StartOperationResultAsync:
+        workflow_id = str(uuid.uuid4())
+        wf_handle = await temporalio.nexus.handler.start_workflow(
+            ctx,
+            WorkflowStartedByNexusOperation.run,
+            input,
+            id=workflow_id,
+        )
+        # TODO(dan)
+        token = temporalio.nexus.token.WorkflowOperationToken.from_workflow_handle(
+            wf_handle
+        ).encode()
+        return StartOperationResultAsync(token)
+
+    async def cancel(self, ctx: CancelOperationContext, token: str) -> None:
+        return await temporalio.nexus.handler.cancel_workflow(ctx, token)
+
+    async def fetch_info(
+        self, ctx: FetchOperationInfoContext, token: str
+    ) -> OperationInfo:
+        raise NotImplementedError(
+            "fetch_info is not supported when a Nexus operation is called by a Temporal workflow"
+        )
+
+    async def fetch_result(
+        self, ctx: FetchOperationResultContext, token: str
+    ) -> MyOutput:
+        raise NotImplementedError(
+            "fetch_result is not supported when a Nexus operation is called by a Temporal workflow, "
+            "but this sample does not demonstrate result fetching"
+        )