RTU: Use nexusrpc.get_service_definition

Author: dandavison

temporalio/workflow.py

@@ -5170,8 +5170,7 @@ def __init__(
         # class.
         if isinstance(service, str):
             self._service_name = service
-        # TODO(preview): make double-underscore attrs private to nexusrpc and expose getters/setters
-        elif service_defn := getattr(service, "__nexus_service__", None):
+        elif service_defn := nexusrpc.get_service_definition(service):
             self._service_name = service_defn.name
         else:
             raise ValueError(

tests/nexus/test_dynamic_creation_of_user_handler_classes.py

@@ -66,7 +66,8 @@ async def test_dynamic_creation_of_user_handler_classes(client: Client):
         )
     )
 
-    service_name = service_cls.__nexus_service__.name
+    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(

tests/nexus/test_handler_operation_definitions.py

@@ -88,11 +88,9 @@ async def workflow_run_operation_with_name_override(
 async def test_collected_operation_names(
     test_case: Type[_TestCase],
 ):
-    service: nexusrpc.ServiceDefinition = getattr(
-        test_case.Service, "__nexus_service__"
-    )
-    assert isinstance(service, nexusrpc.ServiceDefinition)
-    assert service.name == "Service"
+    service_defn = nexusrpc.get_service_definition(test_case.Service)
+    assert isinstance(service_defn, nexusrpc.ServiceDefinition)
+    assert service_defn.name == "Service"
     for method_name, expected_op in test_case.expected_operations.items():
         _, actual_op = nexusrpc.handler.get_operation_factory(
             getattr(test_case.Service, method_name)

tests/nexus/test_workflow_run_operation.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass
 from typing import Any, Type
 
+import nexusrpc
 import pytest
 from nexusrpc import Operation, service
 from nexusrpc.handler import (
@@ -103,10 +104,11 @@ async def test_workflow_run_operation(
 ):
     task_queue = str(uuid.uuid4())
     endpoint = (await create_nexus_endpoint(task_queue, env.client)).endpoint.id
+    assert (service_defn := nexusrpc.get_service_definition(service_handler_cls))
     service_client = ServiceClient(
         server_address=server_address(env),
         endpoint=endpoint,
-        service=service_handler_cls.__nexus_service__.name,
+        service=service_defn.name,
     )
     async with Worker(
         env.client,