Remove top-level aliases

Author: dandavison

temporalio/nexus/handler.py

@@ -56,28 +56,28 @@
 # TODO(dan): confirm approach here: Temporal Nexus services will use this instead of
 # nexusrpc.handler.Operation in order to avoid having to implement fetch_info and
 # fetch_result.
-class Operation(nexusrpc.OperationHandler[I, O]):
+class Operation(nexusrpc.handler.OperationHandler[I, O]):
     """
     Interface that must be implemented by an operation in a Temporal Nexus service.
     """
 
     # fetch_info and fetch_result are not currently to be implemented by Temporal Nexus services.
 
     async def fetch_info(
-        self, token: str, ctx: nexusrpc.FetchOperationInfoContext
-    ) -> nexusrpc.OperationInfo:
+        self, token: str, ctx: nexusrpc.handler.FetchOperationInfoContext
+    ) -> nexusrpc.handler.OperationInfo:
         raise NotImplementedError
 
     async def fetch_result(
-        self, token: str, ctx: nexusrpc.FetchOperationResultContext
+        self, token: str, ctx: nexusrpc.handler.FetchOperationResultContext
     ) -> O:
         raise NotImplementedError
 
 
 # TODO(dan): naming, visibility, make this less awkward
 def get_input_and_output_types_from_workflow_run_start_method(
     start_method: Callable[
-        [S, nexusrpc.StartOperationContext, I],
+        [S, nexusrpc.handler.StartOperationContext, I],
         Awaitable[WorkflowHandle[Any, O]],
     ],
 ) -> tuple[
@@ -120,7 +120,7 @@ def get_input_and_output_types_from_workflow_run_start_method(
 # No-param overload
 @overload
 async def start_workflow(
-    ctx: nexusrpc.StartOperationContext,
+    ctx: nexusrpc.handler.StartOperationContext,
     workflow: MethodAsyncNoParam[SelfType, ReturnType],
     *,
     id: str,
@@ -132,7 +132,7 @@ async def start_workflow(
 # Single-param overload
 @overload
 async def start_workflow(
-    ctx: nexusrpc.StartOperationContext,
+    ctx: nexusrpc.handler.StartOperationContext,
     workflow: MethodAsyncSingleParam[SelfType, ParamType, ReturnType],
     arg: ParamType,
     *,
@@ -145,7 +145,7 @@ async def start_workflow(
 # Multiple-params overload
 @overload
 async def start_workflow(
-    ctx: nexusrpc.StartOperationContext,
+    ctx: nexusrpc.handler.StartOperationContext,
     workflow: Callable[Concatenate[SelfType, MultiParamSpec], Awaitable[ReturnType]],
     *,
     args: Sequence[Any],
@@ -159,7 +159,7 @@ async def start_workflow(
 
 
 async def start_workflow(
-    ctx: nexusrpc.StartOperationContext,
+    ctx: nexusrpc.handler.StartOperationContext,
     workflow: Callable[..., Awaitable[Any]],
     arg: Any = temporalio.common._arg_unset,
     *,
@@ -268,11 +268,11 @@ async def start_workflow(
 # TODO(dan): Not for merge: this is not required for Temporal Nexus, but implementing in
 # order to check that the design extends well to this.
 async def fetch_workflow_info(
-    ctx: nexusrpc.FetchOperationInfoContext,
+    ctx: nexusrpc.handler.FetchOperationInfoContext,
     token: str,
-) -> nexusrpc.OperationInfo:
+) -> nexusrpc.handler.OperationInfo:
     # TODO(dan)
-    return nexusrpc.OperationInfo(
+    return nexusrpc.handler.OperationInfo(
         token=token,
         status=nexusrpc.handler.OperationState.RUNNING,
     )
@@ -281,7 +281,7 @@ async def fetch_workflow_info(
 # TODO(dan): Not for merge: this is not required for Temporal Nexus, but implementing temporarily in
 # order to check that the design extends well to this.
 async def fetch_workflow_result(
-    ctx: nexusrpc.FetchOperationResultContext,
+    ctx: nexusrpc.handler.FetchOperationResultContext,
     token: str,
     client: Optional[Client] = None,
 ) -> Any:
@@ -292,7 +292,7 @@ async def fetch_workflow_result(
 
 
 async def cancel_workflow(
-    ctx: nexusrpc.CancelOperationContext,
+    ctx: nexusrpc.handler.CancelOperationContext,
     token: str,
     client: Optional[Client] = None,
 ) -> None:
@@ -330,12 +330,12 @@ def get_task_queue() -> str:
     return context.task_queue
 
 
-class WorkflowRunOperation(nexusrpc.OperationHandler[I, O], Generic[I, O, S]):
+class WorkflowRunOperation(nexusrpc.handler.OperationHandler[I, O], Generic[I, O, S]):
     def __init__(
         self,
         service: S,
         start_method: Callable[
-            [S, nexusrpc.StartOperationContext, I],
+            [S, nexusrpc.handler.StartOperationContext, I],
             Awaitable[WorkflowHandle[Any, O]],
         ],
         output_type: Optional[Type] = None,
@@ -346,16 +346,16 @@ def __init__(
         # TODO(dan): What is @wraps doing exactly?
         @wraps(start_method)
         async def start(
-            self, ctx: nexusrpc.StartOperationContext, input: I
-        ) -> nexusrpc.StartOperationResultAsync:
+            self, ctx: nexusrpc.handler.StartOperationContext, input: I
+        ) -> nexusrpc.handler.StartOperationResultAsync:
             wf_handle = await start_method(service, ctx, input)
             token = WorkflowOperationToken.from_workflow_handle(wf_handle).encode()
-            return nexusrpc.StartOperationResultAsync(token)
+            return nexusrpc.handler.StartOperationResultAsync(token)
 
         # TODO(dan): get rid of first parameter?
         # TODO(dan): remove before merge; implementing temporarily to check that design extends well to this
         async def fetch_result(
-            self, ctx: nexusrpc.FetchOperationResultContext, token: str
+            self, ctx: nexusrpc.handler.FetchOperationResultContext, token: str
         ) -> O:
             return await fetch_workflow_result(ctx, token)
 
@@ -365,13 +365,13 @@ async def fetch_result(
         self.start = types.MethodType(start, self)
         self.fetch_result = types.MethodType(fetch_result, self)
 
-    async def cancel(self, ctx: nexusrpc.CancelOperationContext, token: str) -> None:
+    async def cancel(self, ctx: nexusrpc.handler.CancelOperationContext, token: str) -> None:
         await cancel_workflow(ctx, token)
 
     # TODO(dan): remove before merge; implementing temporarily to check that design extends well to this
     async def fetch_info(
-        self, ctx: nexusrpc.FetchOperationInfoContext, token: str
-    ) -> nexusrpc.OperationInfo:
+        self, ctx: nexusrpc.handler.FetchOperationInfoContext, token: str
+    ) -> nexusrpc.handler.OperationInfo:
         return await fetch_workflow_info(ctx, token)
 
 
@@ -381,7 +381,7 @@ async def fetch_info(
 def workflow_run_operation(
     start_method: Optional[
         Callable[
-            [S, nexusrpc.StartOperationContext, I],
+            [S, nexusrpc.handler.StartOperationContext, I],
             Awaitable[WorkflowHandle[Any, O]],
         ]
     ] = None,
@@ -392,7 +392,7 @@ def workflow_run_operation(
     Callable[
         [
             Callable[
-                [S, nexusrpc.StartOperationContext, I],
+                [S, nexusrpc.handler.StartOperationContext, I],
                 Awaitable[WorkflowHandle[Any, O]],
             ]
         ],
@@ -401,7 +401,7 @@ def workflow_run_operation(
 ]:
     def decorator(
         start_method: Callable[
-            [S, nexusrpc.StartOperationContext, I],
+            [S, nexusrpc.handler.StartOperationContext, I],
             Awaitable[WorkflowHandle[Any, O]],
         ],
     ) -> Callable[[S], WorkflowRunOperation[I, O, S]]:

temporalio/worker/_nexus.py

@@ -233,14 +233,14 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                 )
             )
             try:
-                ctx = nexusrpc.StartOperationContext(
+                ctx = nexusrpc.handler.StartOperationContext(
                     service=start_request.service,
                     operation=start_request.operation,
                     headers=header,
                     request_id=start_request.request_id,
                     callback_url=start_request.callback,
                     caller_links=[
-                        nexusrpc.Link(url=l.url, type=l.type)
+                        nexusrpc.handler.Link(url=l.url, type=l.type)
                         for l in start_request.links
                     ],
                     callback_header=dict(start_request.callback_header),
@@ -263,9 +263,9 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                         ),
                     )
                 except Exception as err:
-                    raise nexusrpc.HandlerError(
+                    raise nexusrpc.handler.HandlerError(
                         "Data converter failed to decode Nexus operation input",
-                        type=nexusrpc.HandlerErrorType.BAD_REQUEST,
+                        type=nexusrpc.handler.HandlerErrorType.BAD_REQUEST,
                         cause=err,
                         retryable=False,
                     ) from err
@@ -283,7 +283,7 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                     # temporary hack to allow developing type signatures that support
                     # non-async start methods.
                     result = operation_handler.start(ctx, input)
-            except nexusrpc.OperationError as err:
+            except nexusrpc.handler.OperationError as err:
                 return temporalio.bridge.proto.nexus.NexusTaskCompletion(
                     task_token=task_token,
                     completed=temporalio.api.nexus.v1.Response(
@@ -309,7 +309,7 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                     ),
                 )
             else:
-                if isinstance(result, nexusrpc.StartOperationResultAsync):
+                if isinstance(result, nexusrpc.handler.StartOperationResultAsync):
                     op_resp = temporalio.api.nexus.v1.StartOperationResponse(
                         async_success=temporalio.api.nexus.v1.StartOperationResponse.Async(
                             operation_token=result.token,
@@ -319,7 +319,7 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                             ],
                         )
                     )
-                elif isinstance(result, nexusrpc.StartOperationResultSync):
+                elif isinstance(result, nexusrpc.handler.StartOperationResultSync):
                     # TODO(dan): error handling here; what error type should it be?
                     [payload] = await self._data_converter.encode([result.value])
                     op_resp = temporalio.api.nexus.v1.StartOperationResponse(
@@ -331,8 +331,8 @@ async def run() -> temporalio.bridge.proto.nexus.NexusTaskCompletion:
                     # TODO(dan): what should the error response be when the user has failed to wrap their return type?
                     # TODO(dan): unify this failure completion with the path above
                     err = TypeError(
-                        "Operation start method must return either nexusrpc.StartOperationResultSync "
-                        "or nexusrpc.StartOperationResultAsync"
+                        "Operation start method must return either nexusrpc.handler.StartOperationResultSync "
+                        "or nexusrpc.handler.StartOperationResultAsync"
                     )
                     handler_err = _exception_to_handler_error(err)
                     return temporalio.bridge.proto.nexus.NexusTaskCompletion(
@@ -382,7 +382,7 @@ async def _handle_cancel_operation(
                 operation=request.operation,
             )
         )
-        ctx = nexusrpc.CancelOperationContext(
+        ctx = nexusrpc.handler.CancelOperationContext(
             service=request.service,
             operation=request.operation,
         )
@@ -448,7 +448,7 @@ async def _exception_to_failure_proto(
 
     async def _operation_error_to_proto(
         self,
-        err: nexusrpc.OperationError,
+        err: nexusrpc.handler.OperationError,
     ) -> temporalio.api.nexus.v1.UnsuccessfulOperationError:
         cause = err.__cause__
         if cause is None:
@@ -459,7 +459,7 @@ async def _operation_error_to_proto(
         )
 
     async def _handler_error_to_proto(
-        self, err: nexusrpc.HandlerError
+        self, err: nexusrpc.handler.HandlerError
     ) -> temporalio.api.nexus.v1.HandlerError:
         # message HandlerError {
         #     string error_type = 1;
@@ -479,40 +479,40 @@ async def _handler_error_to_proto(
 
 
 # TODO(dan): tests for this function
-def _exception_to_handler_error(err: BaseException) -> nexusrpc.HandlerError:
+def _exception_to_handler_error(err: BaseException) -> nexusrpc.handler.HandlerError:
     # Based on sdk-typescript's convertKnownErrors:
     # https://github.com/temporalio/sdk-typescript/blob/nexus/packages/worker/src/nexus.ts
-    if isinstance(err, nexusrpc.HandlerError):
+    if isinstance(err, nexusrpc.handler.HandlerError):
         print(f"🌈 retryable = {err.retryable}")
         return err
     elif isinstance(err, ApplicationError):
-        return nexusrpc.HandlerError(
+        return nexusrpc.handler.HandlerError(
             # TODO(dan): what should message be?
             err.message,
-            type=nexusrpc.HandlerErrorType.INTERNAL,
+            type=nexusrpc.handler.HandlerErrorType.INTERNAL,
             cause=err,
             # TODO(dan): is there a reason to support retryable=None?
             retryable=not err.non_retryable,
         )
     elif isinstance(err, RPCError):
         if err.status == RPCStatusCode.INVALID_ARGUMENT:
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.BAD_REQUEST, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.BAD_REQUEST, cause=err
             )
         elif err.status in [
             RPCStatusCode.ALREADY_EXISTS,
             RPCStatusCode.FAILED_PRECONDITION,
             RPCStatusCode.OUT_OF_RANGE,
         ]:
-            return nexusrpc.HandlerError(
+            return nexusrpc.handler.HandlerError(
                 err.message,
-                type=nexusrpc.HandlerErrorType.INTERNAL,
+                type=nexusrpc.handler.HandlerErrorType.INTERNAL,
                 cause=err,
                 retryable=False,
             )
         elif err.status in [RPCStatusCode.ABORTED, RPCStatusCode.UNAVAILABLE]:
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.UNAVAILABLE, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.UNAVAILABLE, cause=err
             )
         elif err.status in [
             RPCStatusCode.CANCELLED,
@@ -526,33 +526,33 @@ def _exception_to_handler_error(err: BaseException) -> nexusrpc.HandlerError:
             # we convert to internal because this is not a client auth error and happens
             # when the handler fails to auth with Temporal and should be considered
             # retryable.
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.INTERNAL, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.INTERNAL, cause=err
             )
         elif err.status == RPCStatusCode.NOT_FOUND:
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.NOT_FOUND, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.NOT_FOUND, cause=err
             )
         elif err.status == RPCStatusCode.RESOURCE_EXHAUSTED:
-            return nexusrpc.HandlerError(
+            return nexusrpc.handler.HandlerError(
                 err.message,
-                type=nexusrpc.HandlerErrorType.RESOURCE_EXHAUSTED,
+                type=nexusrpc.handler.HandlerErrorType.RESOURCE_EXHAUSTED,
                 cause=err,
             )
         elif err.status == RPCStatusCode.UNIMPLEMENTED:
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.NOT_IMPLEMENTED, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.NOT_IMPLEMENTED, cause=err
             )
         elif err.status == RPCStatusCode.DEADLINE_EXCEEDED:
-            return nexusrpc.HandlerError(
-                err.message, type=nexusrpc.HandlerErrorType.UPSTREAM_TIMEOUT, cause=err
+            return nexusrpc.handler.HandlerError(
+                err.message, type=nexusrpc.handler.HandlerErrorType.UPSTREAM_TIMEOUT, cause=err
             )
         else:
-            return nexusrpc.HandlerError(
+            return nexusrpc.handler.HandlerError(
                 f"Unhandled RPC error status: {err.status}",
-                type=nexusrpc.HandlerErrorType.INTERNAL,
+                type=nexusrpc.handler.HandlerErrorType.INTERNAL,
                 cause=err,
             )
-    return nexusrpc.HandlerError(
-        str(err), type=nexusrpc.HandlerErrorType.INTERNAL, cause=err
+    return nexusrpc.handler.HandlerError(
+        str(err), type=nexusrpc.handler.HandlerErrorType.INTERNAL, cause=err
     )

temporalio/worker/_worker.py

@@ -159,7 +159,7 @@ def __init__(
                 :py:func:`@activity.defn<temporalio.activity.defn>`. Activities
                 may be async functions or non-async functions.
             nexus_services: Nexus service instances decorated with
-                :py:func:`@nexusrpc.service_handler<nexusrpc.service_handler>`.
+                :py:func:`@nexusrpc.handler.service<nexusrpc.service>`.
             workflows: Workflow classes decorated with
                 :py:func:`@workflow.defn<temporalio.workflow.defn>`.
             activity_executor: Concurrent executor to use for non-async

temporalio/workflow.py

@@ -5182,7 +5182,7 @@ def __init__(
         else:
             raise ValueError(
                 f"`service` may be a name (str), or a class decorated with either "
-                f"@nexusrpc.service_handler or @nexusrpc.contract.service. "
+                f"@nexusrpc.handler.service or @nexusrpc.contract.service. "
                 f"Invalid service type: {type(service)}"
             )
         print(f"🌈 NexusClient using service name: {self._service_name}")