Use getters/setters

Author: dandavison

temporalio/nexus/_decorators.py

@@ -28,6 +28,7 @@
 from temporalio.nexus._util import (
     get_callable_name,
     get_workflow_run_start_method_input_and_output_type_annotations,
+    set_operation_factory,
 )
 
 ServiceHandlerT = TypeVar("ServiceHandlerT")
@@ -124,15 +125,17 @@ async def _start(
             return WorkflowRunOperationHandler(_start, input_type, output_type)
 
         method_name = get_callable_name(start)
-        # TODO(nexus-preview): make double-underscore attrs private to nexusrpc and expose getters/setters
-        operation_handler_factory.__nexus_operation__ = nexusrpc.Operation(
-            name=name or method_name,
-            method_name=method_name,
-            input_type=input_type,
-            output_type=output_type,
+        nexusrpc.set_operation_definition(
+            operation_handler_factory,
+            nexusrpc.Operation(
+                name=name or method_name,
+                method_name=method_name,
+                input_type=input_type,
+                output_type=output_type,
+            ),
         )
 
-        start.__nexus_operation_factory__ = operation_handler_factory
+        set_operation_factory(start, operation_handler_factory)
         return start
 
     if start is None:

temporalio/nexus/_util.py

@@ -14,6 +14,7 @@
     Union,
 )
 
+import nexusrpc
 from nexusrpc import (
     InputT,
     OutputT,
@@ -129,19 +130,19 @@ def get_callable_name(fn: Callable[..., Any]) -> str:
 def get_operation_factory(
     obj: Any,
 ) -> tuple[
-    Optional[Callable[[Any], OperationHandler[InputT, OutputT]]],
-    Optional[nexusrpc.Operation[InputT, OutputT]],
+    Optional[Callable[[Any], Any]],
+    Optional[nexusrpc.Operation[Any, Any]],
 ]:
     """Return the :py:class:`Operation` for the object along with the factory function.
 
     ``obj`` should be a decorated operation start method.
     """
-    op_defn = get_operation_definition(obj)
+    op_defn = nexusrpc.get_operation_definition(obj)
     if op_defn:
         factory = obj
     else:
         if factory := getattr(obj, "__nexus_operation_factory__", None):
-            op_defn = get_operation_definition(factory)
+            op_defn = nexusrpc.get_operation_definition(factory)
     if not isinstance(op_defn, nexusrpc.Operation):
         return None, None
     return factory, op_defn
@@ -150,7 +151,7 @@ def get_operation_factory(
 # TODO(nexus-preview) Copied from nexusrpc
 def set_operation_factory(
     obj: Any,
-    operation_factory: Callable[[Any], OperationHandler[InputT, OutputT]],
+    operation_factory: Callable[[Any], Any],
 ) -> None:
     """Set the :py:class:`OperationHandler` factory for this object.