diff --git a/README.md b/README.md index 16426399..2954d242 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,7 @@ Some examples require extra dependencies. See each sample's directory for specif This contains two samples, one sending messages to an existing workflow and a second that creates a workflow through Nexus and sends messages to it. * [nexus_multiple_args](nexus_multiple_args) - Map a Nexus operation to a handler workflow that takes multiple arguments. +* [nexus_standalone_activity](nexus_standalone_activity) - Back a Nexus operation with a standalone Activity. * [nexus_standalone_operations](nexus_standalone_operations) - Execute Nexus operations directly from client code, without wrapping them in a workflow. * [open_telemetry](open_telemetry) - Trace workflows with OpenTelemetry. diff --git a/nexus_standalone_activity/README.md b/nexus_standalone_activity/README.md new file mode 100644 index 00000000..1f795424 --- /dev/null +++ b/nexus_standalone_activity/README.md @@ -0,0 +1,59 @@ +# Nexus operation backed by a standalone Activity + +This sample shows how to implement a `TemporalOperationHandler` that starts a +standalone Activity as the backing execution for a Nexus operation. When the Activity +finishes, Temporal delivers its result to the Nexus caller. The default handler +cancellation implementation also forwards Nexus cancellation to the Activity. + +The APIs used by this sample are experimental and may change incompatibly. + +### Sample structure + +- [service.py](./service.py) defines the Nexus service shared by caller and handler. +- [activity.py](./activity.py) defines the standalone Activity. +- [handler.py](./handler.py) implements `TemporalOperationHandler.start_operation`. +- [worker.py](./worker.py) hosts the Nexus handler and Activity. +- [starter.py](./starter.py) executes the Nexus operation from client code. + +## Run locally + +This sample requires the [Temporal dev server build that supports standalone Nexus operations](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support) and Activity +callbacks enabled. + +1. Start the server with caller and handler namespaces: + + ```bash + ./temporal server start-dev \ + --dynamic-config-value activity.enableCallbacks=true \ + --namespace nexus-standalone-activity-caller \ + --namespace nexus-standalone-activity-handler + ``` + +2. Create an endpoint targeting the handler namespace and task queue: + + ```bash + ./temporal operator nexus endpoint create \ + --name nexus-standalone-activity-endpoint \ + --target-namespace nexus-standalone-activity-handler \ + --target-task-queue nexus-standalone-activity-handler + ``` + +3. Start the handler Worker: + + ```bash + TEMPORAL_NAMESPACE=nexus-standalone-activity-handler \ + uv run nexus_standalone_activity/worker.py + ``` + +4. Execute the operation from the caller namespace: + + ```bash + TEMPORAL_NAMESPACE=nexus-standalone-activity-caller \ + uv run nexus_standalone_activity/starter.py + ``` + +Expected output: + +```text +Hello, World! +``` diff --git a/nexus_standalone_activity/__init__.py b/nexus_standalone_activity/__init__.py new file mode 100644 index 00000000..6dca48d1 --- /dev/null +++ b/nexus_standalone_activity/__init__.py @@ -0,0 +1 @@ +"""Nexus operation backed by a standalone Activity sample.""" diff --git a/nexus_standalone_activity/activity.py b/nexus_standalone_activity/activity.py new file mode 100644 index 00000000..a2ed8a50 --- /dev/null +++ b/nexus_standalone_activity/activity.py @@ -0,0 +1,10 @@ +"""Activity used as the backing execution for the Nexus operation.""" + +from temporalio import activity + +from nexus_standalone_activity.service import GreetingInput, GreetingOutput + + +@activity.defn +async def create_greeting(input: GreetingInput) -> GreetingOutput: + return GreetingOutput(message=f"Hello, {input.name}!") diff --git a/nexus_standalone_activity/handler.py b/nexus_standalone_activity/handler.py new file mode 100644 index 00000000..960d75d0 --- /dev/null +++ b/nexus_standalone_activity/handler.py @@ -0,0 +1,36 @@ +"""Temporal operation handler that starts a standalone Activity.""" + +from datetime import timedelta + +import nexusrpc.handler +from temporalio import nexus + +from nexus_standalone_activity.activity import create_greeting +from nexus_standalone_activity.service import ( + GreetingInput, + GreetingOutput, + GreetingService, +) + + +def get_activity_id(input: GreetingInput) -> str: + return f"greeting-{input.name}" + + +@nexusrpc.handler.service_handler(service=GreetingService) +class GreetingServiceHandler: + @nexus.temporal_operation + async def greet( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: GreetingInput, + ) -> nexus.TemporalOperationResult[GreetingOutput]: + # The standalone Activity becomes the asynchronous backing execution for + # this Nexus operation. Omitting task_queue uses the Nexus Worker's queue. + return await client.start_activity( + create_greeting, + input, + id=get_activity_id(input), + start_to_close_timeout=timedelta(seconds=10), + ) diff --git a/nexus_standalone_activity/service.py b/nexus_standalone_activity/service.py new file mode 100644 index 00000000..ad3dad25 --- /dev/null +++ b/nexus_standalone_activity/service.py @@ -0,0 +1,20 @@ +"""Nexus service definition shared by the caller and handler.""" + +from dataclasses import dataclass + +import nexusrpc + + +@dataclass +class GreetingInput: + name: str + + +@dataclass +class GreetingOutput: + message: str + + +@nexusrpc.service +class GreetingService: + greet: nexusrpc.Operation[GreetingInput, GreetingOutput] diff --git a/nexus_standalone_activity/starter.py b/nexus_standalone_activity/starter.py new file mode 100644 index 00000000..1db47e11 --- /dev/null +++ b/nexus_standalone_activity/starter.py @@ -0,0 +1,34 @@ +"""Client that executes the activity-backed Nexus operation.""" + +import asyncio +import uuid +from datetime import timedelta + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig + +from nexus_standalone_activity.service import GreetingInput, GreetingService + +ENDPOINT_NAME = "nexus-standalone-activity-endpoint" + + +async def main() -> None: + config = ClientConfig.load_client_connect_config() + _ = config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + + nexus_client = client.create_nexus_client( + service=GreetingService, + endpoint=ENDPOINT_NAME, + ) + result = await nexus_client.execute_operation( + GreetingService.greet, + GreetingInput(name="World"), + id=f"greeting-{uuid.uuid4()}", + schedule_to_close_timeout=timedelta(seconds=10), + ) + print(result.message) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/nexus_standalone_activity/worker.py b/nexus_standalone_activity/worker.py new file mode 100644 index 00000000..e4a15a0f --- /dev/null +++ b/nexus_standalone_activity/worker.py @@ -0,0 +1,41 @@ +"""Worker hosting the Nexus handler and its standalone Activity.""" + +import asyncio +import logging + +from temporalio.client import Client +from temporalio.envconfig import ClientConfig +from temporalio.worker import Worker + +from nexus_standalone_activity.activity import create_greeting +from nexus_standalone_activity.handler import GreetingServiceHandler + +TASK_QUEUE = "nexus-standalone-activity-handler" + +interrupt_event = asyncio.Event() + + +async def main() -> None: + logging.basicConfig(level=logging.INFO) + + config = ClientConfig.load_client_connect_config() + _ = config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**config) + + async with Worker( + client, + task_queue=TASK_QUEUE, + activities=[create_greeting], + nexus_service_handlers=[GreetingServiceHandler()], + ): + logging.info("Worker started, ctrl+c to exit") + _ = await interrupt_event.wait() + + +if __name__ == "__main__": + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(main()) + except KeyboardInterrupt: + interrupt_event.set() + loop.run_until_complete(loop.shutdown_asyncgens()) diff --git a/tests/conftest.py b/tests/conftest.py index b857e0d9..7f7fe066 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -42,6 +42,10 @@ async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]: env_type = request.config.getoption("--workflow-environment") if env_type == "local": env = await WorkflowEnvironment.start_local( + dev_server_extra_args=[ + "--dynamic-config-value", + "activity.enableCallbacks=true", + ], dev_server_download_version="v1.7.4-standalone-nexus-operations", ) elif env_type == "time-skipping": diff --git a/tests/nexus_standalone_activity/__init__.py b/tests/nexus_standalone_activity/__init__.py new file mode 100644 index 00000000..7d425474 --- /dev/null +++ b/tests/nexus_standalone_activity/__init__.py @@ -0,0 +1 @@ +"""Tests for the Nexus standalone Activity sample.""" diff --git a/tests/nexus_standalone_activity/nexus_standalone_activity_test.py b/tests/nexus_standalone_activity/nexus_standalone_activity_test.py new file mode 100644 index 00000000..e146c960 --- /dev/null +++ b/tests/nexus_standalone_activity/nexus_standalone_activity_test.py @@ -0,0 +1,57 @@ +import uuid +from datetime import timedelta + +import pytest +from temporalio.client import Client +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker + +from nexus_standalone_activity.activity import create_greeting +from nexus_standalone_activity.handler import GreetingServiceHandler +from nexus_standalone_activity.service import ( + GreetingInput, + GreetingOutput, + GreetingService, +) +from nexus_standalone_activity.worker import TASK_QUEUE +from tests.helpers.nexus import create_nexus_endpoint, delete_nexus_endpoint + + +async def test_nexus_operation_backed_by_standalone_activity( + client: Client, env: WorkflowEnvironment +) -> None: + if env.supports_time_skipping: + pytest.skip("Time-skipping server does not support standalone Nexus operations") + + endpoint_name = f"test-nexus-standalone-activity-{uuid.uuid4()}" + create_response = await create_nexus_endpoint( + name=endpoint_name, + task_queue=TASK_QUEUE, + client=client, + ) + try: + async with Worker( + client, + task_queue=TASK_QUEUE, + activities=[create_greeting], + nexus_service_handlers=[GreetingServiceHandler()], + ): + nexus_client = client.create_nexus_client( + service=GreetingService, + endpoint=endpoint_name, + ) + result = await nexus_client.execute_operation( + GreetingService.greet, + GreetingInput(name="Test"), + id=str(uuid.uuid4()), + schedule_to_close_timeout=timedelta(seconds=10), + ) + + assert isinstance(result, GreetingOutput) + assert result.message == "Hello, Test!" + finally: + _ = await delete_nexus_endpoint( + id=create_response.endpoint.id, + version=create_response.endpoint.version, + client=client, + )