Cleanup

Author: dandavison

temporalio/nexus/handler.py

@@ -264,8 +264,8 @@ def factory(
                 f"expected {start_method} to be a function or callable instance."
             )
 
-        factory.__nexus_operation__ = nexusrpc.Operation._create(
-            name=name,
+        factory.__nexus_operation__ = nexusrpc.Operation(
+            name=name or method_name,
             method_name=method_name,
             input_type=input_type,
             output_type=output_type,

temporalio/worker/_interceptor.py

@@ -334,6 +334,7 @@ def __post_init__(self) -> None:
     def operation_name(self) -> str:
         return self._operation_name
 
+    # TODO(nexus-prerelease) contravariant type in output
     @property
     def input_type(self) -> Optional[Type[InputT]]:
         return self._input_type

temporalio/worker/_nexus.py

@@ -178,7 +178,7 @@ async def _handle_cancel_operation_task(
         temporalio.nexus.current_context.set(
             temporalio.nexus.Context(operation_context=ctx)
         )
-        # TODO(nexus-prerelease): header
+        # TODO(nexus-prerelease): headers
         try:
             await self._handler.cancel_operation(ctx, request.operation_token)
         except Exception as err:
@@ -219,6 +219,7 @@ async def _handle_start_operation_task(
 
         try:
             start_response = await self._start_operation(start_request, headers)
+        # TODO(nexus-prerelease): handle BrokenExecutor by failing the worker
         except BaseException as err:
             handler_err = _exception_to_handler_error(err)
             completion = temporalio.bridge.proto.nexus.NexusTaskCompletion(
@@ -235,7 +236,6 @@ async def _handle_start_operation_task(
 
         try:
             await self._bridge_worker().complete_nexus_task(completion)
-        # TODO(nexus-prerelease): handle BrokenExecutor by failing the worker
         except Exception:
             temporalio.nexus.logger.exception("Failed to send Nexus task completion")
         finally:
@@ -290,8 +290,8 @@ async def _start_operation(
                     async_success=temporalio.api.nexus.v1.StartOperationResponse.Async(
                         operation_token=result.token,
                         links=[
-                            temporalio.api.nexus.v1.Link(url=l.url, type=l.type)
-                            for l in ctx.outbound_links
+                            temporalio.api.nexus.v1.Link(url=link.url, type=link.type)
+                            for link in ctx.outbound_links
                         ],
                     )
                 )
@@ -305,8 +305,9 @@ async def _start_operation(
             else:
                 raise _exception_to_handler_error(
                     TypeError(
-                        "Operation start method must return either nexusrpc.handler.StartOperationResultSync "
-                        "or nexusrpc.handler.StartOperationResultAsync"
+                        "Operation start method must return either "
+                        "nexusrpc.handler.StartOperationResultSync or "
+                        "nexusrpc.handler.StartOperationResultAsync."
                     )
                 )
         except nexusrpc.handler.OperationError as err:
@@ -321,7 +322,6 @@ async def _exception_to_failure_proto(
         api_failure = temporalio.api.failure.v1.Failure()
         await self._data_converter.encode_failure(err, api_failure)
         api_failure = google.protobuf.json_format.MessageToDict(api_failure)
-        # TODO(nexus-prerelease): is metadata correct and playing intended role here?
         return temporalio.api.nexus.v1.Failure(
             message=api_failure.pop("message", ""),
             metadata={"type": "temporal.api.failure.v1.Failure"},
@@ -375,14 +375,14 @@ async def deserialize(
                 [self.payload],
                 type_hints=[as_type] if as_type else None,
             )
+            return input
         except Exception as err:
             raise nexusrpc.handler.HandlerError(
                 "Data converter failed to decode Nexus operation input",
                 type=nexusrpc.handler.HandlerErrorType.BAD_REQUEST,
                 cause=err,
                 retryable=False,
             ) from err
-        return input
 
 
 # TODO(nexus-prerelease): tests for this function

tests/nexus/test_handler.py

@@ -573,7 +573,7 @@ class OperationHandlerReturningUnwrappedResultError(_FailureTestCase):
         retryable_header=False,
         failure_message=(
             "Operation start method must return either "
-            "nexusrpc.handler.StartOperationResultSync or nexusrpc.handler.StartOperationResultAsync"
+            "nexusrpc.handler.StartOperationResultSync or nexusrpc.handler.StartOperationResultAsync."
         ),
     )
 

tests/nexus/test_handler_operation_definitions.py

@@ -38,7 +38,8 @@ async def workflow_run_operation_handler(
         ) -> WorkflowHandle[Any, Output]: ...
 
     expected_operations = {
-        "workflow_run_operation_handler": nexusrpc.Operation._create(
+        "workflow_run_operation_handler": nexusrpc.Operation(
+            name="workflow_run_operation_handler",
             method_name="workflow_run_operation_handler",
             input_type=Input,
             output_type=Output,
@@ -66,7 +67,7 @@ async def workflow_run_operation_with_name_override(
         ) -> WorkflowHandle[Any, Output]: ...
 
     expected_operations = {
-        "workflow_run_operation_with_name_override": nexusrpc.Operation._create(
+        "workflow_run_operation_with_name_override": nexusrpc.Operation(
             name="operation-name",
             method_name="workflow_run_operation_with_name_override",
             input_type=Input,