Copy get_types utility from nexusrpc

Author: dandavison

temporalio/nexus/_util.py

@@ -10,12 +10,10 @@
     Callable,
     Optional,
     Type,
+    Union,
 )
 
-from nexusrpc.handler import (
-    StartOperationContext,
-    get_start_method_input_and_output_type_annotations,
-)
+from nexusrpc.handler import StartOperationContext
 from nexusrpc.types import (
     InputT,
     OutputT,
@@ -41,7 +39,8 @@ def get_workflow_run_start_method_input_and_output_type_annotations(
     `start` must be a type-annotated start method that returns a
     :py:class:`temporalio.nexus.WorkflowHandle`.
     """
-    input_type, output_type = get_start_method_input_and_output_type_annotations(start)
+
+    input_type, output_type = _get_start_method_input_and_output_type_annotations(start)
     origin_type = typing.get_origin(output_type)
     if not origin_type:
         output_type = None
@@ -66,6 +65,51 @@ def get_workflow_run_start_method_input_and_output_type_annotations(
     return input_type, output_type
 
 
+def _get_start_method_input_and_output_type_annotations(
+    start: Callable[
+        [ServiceHandlerT, WorkflowRunOperationContext, InputT],
+        Union[OutputT, Awaitable[OutputT]],
+    ],
+) -> tuple[
+    Optional[Type[InputT]],
+    Optional[Type[OutputT]],
+]:
+    """Return operation input and output types.
+
+    `start` must be a type-annotated start method that returns a synchronous result.
+    """
+    try:
+        type_annotations = typing.get_type_hints(start)
+    except TypeError:
+        # TODO(preview): stacklevel
+        warnings.warn(
+            f"Expected decorated start method {start} to have type annotations"
+        )
+        return None, None
+    output_type = type_annotations.pop("return", None)
+
+    if len(type_annotations) != 2:
+        # TODO(preview): stacklevel
+        suffix = f": {type_annotations}" if type_annotations else ""
+        warnings.warn(
+            f"Expected decorated start method {start} to have exactly 2 "
+            f"type-annotated parameters (ctx and input), but it has {len(type_annotations)}"
+            f"{suffix}."
+        )
+        input_type = None
+    else:
+        ctx_type, input_type = type_annotations.values()
+        if not issubclass(ctx_type, WorkflowRunOperationContext):
+            # TODO(preview): stacklevel
+            warnings.warn(
+                f"Expected first parameter of {start} to be an instance of "
+                f"WorkflowRunOperationContext, but is {ctx_type}."
+            )
+            input_type = None
+
+    return input_type, output_type
+
+
 def get_callable_name(fn: Callable[..., Any]) -> str:
     method_name = getattr(fn, "__name__", None)
     if not method_name and callable(fn) and hasattr(fn, "__call__"):