Move StartWorkflowOperationResult type helper into temporal sdk

Author: dandavison

temporalio/nexus/handler.py

@@ -17,6 +17,7 @@
     Generic,
     Optional,
     Sequence,
+    Type,
     TypeVar,
 )
 
@@ -118,6 +119,31 @@ def to_workflow_handle(token: str, client: Client) -> WorkflowHandle[Any, O]:
         return client.get_workflow_handle(workflow_id, run_id=run_id)
 
 
+def get_input_and_output_types_from_async_start_method(
+    start_method: Callable[
+        [S, I, nexusrpc.handler.StartOperationOptions],
+        Awaitable[StartWorkflowOperationResult[O]],
+    ],
+) -> tuple[Type[I], Type[O]]:
+    input_type, output_type = (
+        nexusrpc.handler.get_input_and_output_types_from_sync_start_method(start_method)
+    )
+    origin_type = typing.get_origin(output_type)
+    if not origin_type or not issubclass(origin_type, StartWorkflowOperationResult):
+        raise TypeError(
+            f"The return type of {start_method.__name__} must be a subclass of StartWorkflowOperationResult, "
+            f"but is {output_type}"
+        )
+
+    args = typing.get_args(output_type)
+    if len(args) != 1:
+        raise TypeError(
+            f"The return type of {start_method.__name__} must have exactly one type parameter, "
+            f"but has {len(args)}: {args}"
+        )
+    return input_type, args[0]
+
+
 # TODO(dan): overloads should use SelfType, ParamType, ReturnType?
 
 
@@ -323,10 +349,8 @@ def workflow_run_operation(
     def factory(service: S) -> WorkflowRunOperation[I, O, S]:
         return WorkflowRunOperation(service, start_method)
 
-    input_type, output_type = (
-        nexusrpc.handler.get_input_and_output_types_from_async_start_method(
-            start_method
-        )
+    input_type, output_type = get_input_and_output_types_from_async_start_method(
+        start_method
     )
     factory.__nexus_operation__ = nexusrpc.handler.NexusOperationDefinition(
         name=start_method.__name__,