Respond to upstream: do not use get_operation_factory

Author: dandavison

temporalio/nexus/_util.py

@@ -13,7 +13,6 @@
     TypeVar,
 )
 
-import nexusrpc
 from nexusrpc import (
     InputT,
     OutputT,
@@ -118,28 +117,6 @@ def get_callable_name(fn: Callable[..., Any]) -> str:
     return method_name
 
 
-# TODO(nexus-preview) Copied from nexusrpc
-def get_operation_factory(
-    obj: Any,
-) -> tuple[
-    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 = nexusrpc.get_operation(obj)
-    if op:
-        factory = obj
-    else:
-        if factory := getattr(obj, "__nexus_operation_factory__", None):
-            op = nexusrpc.get_operation(factory)
-    if not isinstance(op, nexusrpc.Operation):
-        return None, None
-    return factory, op
-
-
 # TODO(nexus-preview) Copied from nexusrpc
 def set_operation_factory(
     obj: Any,

temporalio/worker/_interceptor.py

@@ -299,15 +299,14 @@ class StartNexusOperationInput(Generic[InputT, OutputT]):
     input: InputT
     schedule_to_close_timeout: Optional[timedelta]
     headers: Optional[Mapping[str, str]]
-    output_type: Optional[Type[OutputT]] = None
+    output_type: Optional[type[OutputT]] = None
 
     def __post_init__(self) -> None:
         """Initialize operation-specific attributes after dataclass creation."""
         if isinstance(self.operation, nexusrpc.Operation):
             self.output_type = self.operation.output_type
         elif callable(self.operation):
-            _, op = temporalio.nexus._util.get_operation_factory(self.operation)
-            if isinstance(op, nexusrpc.Operation):
+            if op := nexusrpc.get_operation(self.operation):
                 self.output_type = op.output_type
             else:
                 raise ValueError(
@@ -326,8 +325,7 @@ def operation_name(self) -> str:
         elif isinstance(self.operation, str):
             return self.operation
         elif callable(self.operation):
-            _, op = temporalio.nexus._util.get_operation_factory(self.operation)
-            if isinstance(op, nexusrpc.Operation):
+            if op := nexusrpc.get_operation(self.operation):
                 return op.name
             else:
                 raise ValueError(

tests/nexus/test_dynamic_creation_of_user_handler_classes.py

@@ -3,11 +3,9 @@
 import httpx
 import nexusrpc.handler
 import pytest
-from nexusrpc.handler import sync_operation
 
 from temporalio import nexus, workflow
 from temporalio.client import Client
-from temporalio.nexus._util import get_operation_factory
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 from tests.helpers.nexus import ServiceClient, create_nexus_endpoint
@@ -107,70 +105,3 @@ async def test_run_nexus_service_from_programmatically_created_service_handler(
                 json=1,
             )
             assert response.status_code == 201
-
-
-def make_incrementer_user_service_definition_and_service_handler_classes(
-    op_names: list[str],
-) -> tuple[type, type]:
-    #
-    # service contract
-    #
-
-    ops = {name: nexusrpc.Operation[int, int] for name in op_names}
-    service_cls: type = nexusrpc.service(type("ServiceContract", (), ops))
-
-    #
-    # service handler
-    #
-    @sync_operation
-    async def _increment_op(
-        self,
-        ctx: nexusrpc.handler.StartOperationContext,
-        input: int,
-    ) -> int:
-        return input + 1
-
-    op_handler_factories = {}
-    for name in op_names:
-        op_handler_factory, _ = get_operation_factory(_increment_op)
-        assert op_handler_factory
-        op_handler_factories[name] = op_handler_factory
-
-    handler_cls: type = nexusrpc.handler.service_handler(service=service_cls)(
-        type("ServiceImpl", (), op_handler_factories)
-    )
-
-    return service_cls, handler_cls
-
-
-@pytest.mark.skip(
-    reason="Dynamic creation of service contract using type() is not supported"
-)
-async def test_dynamic_creation_of_user_handler_classes(
-    client: Client, env: WorkflowEnvironment
-):
-    task_queue = str(uuid.uuid4())
-
-    service_cls, handler_cls = (
-        make_incrementer_user_service_definition_and_service_handler_classes(
-            ["increment"]
-        )
-    )
-
-    assert (service_defn := nexusrpc.get_service_definition(service_cls))
-    service_name = service_defn.name
-
-    endpoint = (await create_nexus_endpoint(task_queue, client)).endpoint.id
-    async with Worker(
-        client,
-        task_queue=task_queue,
-        nexus_service_handlers=[handler_cls()],
-    ):
-        server_address = ServiceClient.default_server_address(env)
-        async with httpx.AsyncClient() as http_client:
-            response = await http_client.post(
-                f"http://{server_address}/nexus/endpoints/{endpoint}/services/{service_name}/increment",
-                json=1,
-            )
-            assert response.status_code == 200
-            assert response.json() == 2

tests/nexus/test_handler_operation_definitions.py

@@ -11,7 +11,6 @@
 
 from temporalio import nexus
 from temporalio.nexus import WorkflowRunOperationContext, workflow_run_operation
-from temporalio.nexus._util import get_operation_factory
 
 
 @dataclass
@@ -96,7 +95,7 @@ async def test_collected_operation_names(
     assert isinstance(service_defn, nexusrpc.ServiceDefinition)
     assert service_defn.name == "Service"
     for method_name, expected_op in test_case.expected_operations.items():
-        _, actual_op = get_operation_factory(getattr(test_case.Service, method_name))
+        actual_op = nexusrpc.get_operation(getattr(test_case.Service, method_name))
         assert isinstance(actual_op, nexusrpc.Operation)
         assert actual_op.name == expected_op.name
         assert actual_op.input_type == expected_op.input_type