Workflow: cross-app client operations

dapr/ext/workflow/_durabletask/aio/client.py

@@ -40,6 +40,7 @@
     WorkflowState,
     _TransientTimeout,
     new_orchestration_state,
+    new_task_router,
 )
 
 # If `opentelemetry-instrumentation-grpc` is available, enable the gRPC client interceptor
@@ -114,6 +115,7 @@ async def schedule_new_orchestration(
         instance_id: Optional[str] = None,
         start_at: Optional[datetime] = None,
         reuse_id_policy: Optional[WorkflowIdReusePolicy] = None,
+        app_id: Optional[str] = None,
     ) -> str:
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
 
@@ -125,23 +127,41 @@ async def schedule_new_orchestration(
             else None,
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
             version=helpers.get_string_value(None),
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = await self._get_stub().StartInstance(req)
         return res.instanceId
 
     async def get_orchestration_state(
-        self, instance_id: str, *, fetch_payloads: bool = True
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = True,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         res: pb.GetInstanceResponse = await self._get_stub().GetInstance(req)
         return new_orchestration_state(req.instanceId, res)
 
     async def wait_for_orchestration_start(
-        self, instance_id: str, *, fetch_payloads: bool = False, timeout: Optional[int] = 0
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = False,
+        timeout: Optional[int] = 0,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         self._logger.info(
             f"Waiting {'indefinitely' if timeout in (0, None) else f'up to {timeout}s'} for instance '{instance_id}' to start."
         )
@@ -158,9 +178,18 @@ async def _call(grpc_timeout):
             raise TimeoutError('Timed-out waiting for the orchestration to start')
 
     async def wait_for_orchestration_completion(
-        self, instance_id: str, *, fetch_payloads: bool = True, timeout: Optional[int] = 0
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = True,
+        timeout: Optional[int] = 0,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         self._logger.info(
             f"Waiting {'indefinitely' if timeout in (0, None) else f'up to {timeout}s'} for instance '{instance_id}' to complete."
         )
@@ -270,40 +299,78 @@ async def _call_with_transient_retry(self, instance_id, timeout, call_fn):
                         raise _TransientTimeout()
 
     async def raise_orchestration_event(
-        self, instance_id: str, event_name: str, *, data: Optional[Any] = None
+        self,
+        instance_id: str,
+        event_name: str,
+        *,
+        data: Optional[Any] = None,
+        app_id: Optional[str] = None,
     ):
         req = pb.RaiseEventRequest(
             instanceId=instance_id,
             name=event_name,
             input=wrappers_pb2.StringValue(value=shared.to_json(data)) if data else None,
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.")
         await self._get_stub().RaiseEvent(req)
 
     async def terminate_orchestration(
-        self, instance_id: str, *, output: Optional[Any] = None, recursive: bool = True
+        self,
+        instance_id: str,
+        *,
+        output: Optional[Any] = None,
+        recursive: bool = True,
+        app_id: Optional[str] = None,
     ):
         req = pb.TerminateRequest(
             instanceId=instance_id,
             output=wrappers_pb2.StringValue(value=shared.to_json(output)) if output else None,
             recursive=recursive,
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Terminating instance '{instance_id}'.")
         await self._get_stub().TerminateInstance(req)
 
-    async def suspend_orchestration(self, instance_id: str):
-        req = pb.SuspendRequest(instanceId=instance_id)
+    async def suspend_orchestration(
+        self,
+        instance_id: str,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.SuspendRequest(
+            instanceId=instance_id,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Suspending instance '{instance_id}'.")
         await self._get_stub().SuspendInstance(req)
 
-    async def resume_orchestration(self, instance_id: str):
-        req = pb.ResumeRequest(instanceId=instance_id)
+    async def resume_orchestration(
+        self,
+        instance_id: str,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.ResumeRequest(
+            instanceId=instance_id,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Resuming instance '{instance_id}'.")
         await self._get_stub().ResumeInstance(req)
 
-    async def purge_orchestration(self, instance_id: str, recursive: bool = True):
-        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
+    async def purge_orchestration(
+        self,
+        instance_id: str,
+        recursive: bool = True,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.PurgeInstancesRequest(
+            instanceId=instance_id,
+            recursive=recursive,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Purging instance '{instance_id}'.")
         await self._get_stub().PurgeInstances(req)

dapr/ext/workflow/_durabletask/client.py

@@ -109,6 +109,16 @@ def raise_if_failed(self):
             )
 
 
+def new_task_router(app_id: Optional[str]) -> Optional[pb.TaskRouter]:
+    """Builds a TaskRouter targeting another app, or None when no routing is requested.
+
+    Only the target app ID is set; the sidecar stamps sourceAppID.
+    """
+    if app_id is None:
+        return None
+    return pb.TaskRouter(targetAppID=app_id)
+
+
 class OrchestrationFailedError(Exception):
     def __init__(self, message: str, failure_details: task.FailureDetails):
         super().__init__(message)
@@ -210,6 +220,7 @@ def schedule_new_orchestration(
         instance_id: Optional[str] = None,
         start_at: Optional[datetime] = None,
         reuse_id_policy: Optional[WorkflowIdReusePolicy] = None,
+        app_id: Optional[str] = None,
     ) -> str:
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
 
@@ -223,23 +234,41 @@ def schedule_new_orchestration(
             input=input_pb,
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
             version=wrappers_pb2.StringValue(value=''),
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = self._stub.StartInstance(req)
         return res.instanceId
 
     def get_orchestration_state(
-        self, instance_id: str, *, fetch_payloads: bool = True
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = True,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         res: pb.GetInstanceResponse = self._stub.GetInstance(req)
         return new_orchestration_state(req.instanceId, res)
 
     def wait_for_orchestration_start(
-        self, instance_id: str, *, fetch_payloads: bool = False, timeout: Optional[int] = 0
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = False,
+        timeout: Optional[int] = 0,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         self._logger.info(
             f"Waiting {'indefinitely' if timeout in (0, None) else f'up to {timeout}s'} for instance '{instance_id}' to start."
         )
@@ -254,9 +283,18 @@ def _call(grpc_timeout):
             raise TimeoutError('Timed-out waiting for the orchestration to start')
 
     def wait_for_orchestration_completion(
-        self, instance_id: str, *, fetch_payloads: bool = True, timeout: Optional[int] = 0
+        self,
+        instance_id: str,
+        *,
+        fetch_payloads: bool = True,
+        timeout: Optional[int] = 0,
+        app_id: Optional[str] = None,
     ) -> Optional[WorkflowState]:
-        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        req = pb.GetInstanceRequest(
+            instanceId=instance_id,
+            getInputsAndOutputs=fetch_payloads,
+            router=new_task_router(app_id),
+        )
         self._logger.info(
             f"Waiting {'indefinitely' if timeout in (0, None) else f'up to {timeout}s'} for instance '{instance_id}' to complete."
         )
@@ -393,40 +431,78 @@ def _call_with_transient_retry(self, instance_id, timeout, call_fn):
                         raise _TransientTimeout()
 
     def raise_orchestration_event(
-        self, instance_id: str, event_name: str, *, data: Optional[Any] = None
+        self,
+        instance_id: str,
+        event_name: str,
+        *,
+        data: Optional[Any] = None,
+        app_id: Optional[str] = None,
     ):
         req = pb.RaiseEventRequest(
             instanceId=instance_id,
             name=event_name,
             input=wrappers_pb2.StringValue(value=shared.to_json(data)) if data else None,
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.")
         self._stub.RaiseEvent(req)
 
     def terminate_orchestration(
-        self, instance_id: str, *, output: Optional[Any] = None, recursive: bool = True
+        self,
+        instance_id: str,
+        *,
+        output: Optional[Any] = None,
+        recursive: bool = True,
+        app_id: Optional[str] = None,
     ):
         req = pb.TerminateRequest(
             instanceId=instance_id,
             output=wrappers_pb2.StringValue(value=shared.to_json(output)) if output else None,
             recursive=recursive,
+            router=new_task_router(app_id),
         )
 
         self._logger.info(f"Terminating instance '{instance_id}'.")
         self._stub.TerminateInstance(req)
 
-    def suspend_orchestration(self, instance_id: str):
-        req = pb.SuspendRequest(instanceId=instance_id)
+    def suspend_orchestration(
+        self,
+        instance_id: str,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.SuspendRequest(
+            instanceId=instance_id,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Suspending instance '{instance_id}'.")
         self._stub.SuspendInstance(req)
 
-    def resume_orchestration(self, instance_id: str):
-        req = pb.ResumeRequest(instanceId=instance_id)
+    def resume_orchestration(
+        self,
+        instance_id: str,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.ResumeRequest(
+            instanceId=instance_id,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Resuming instance '{instance_id}'.")
         self._stub.ResumeInstance(req)
 
-    def purge_orchestration(self, instance_id: str, recursive: bool = True):
-        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
+    def purge_orchestration(
+        self,
+        instance_id: str,
+        recursive: bool = True,
+        *,
+        app_id: Optional[str] = None,
+    ):
+        req = pb.PurgeInstancesRequest(
+            instanceId=instance_id,
+            recursive=recursive,
+            router=new_task_router(app_id),
+        )
         self._logger.info(f"Purging instance '{instance_id}'.")
         self._stub.PurgeInstances(req)

dapr/ext/workflow/_durabletask/internal/PROTO_SOURCE_COMMIT_HASH

@@ -1 +1 @@
-9d3681cb82a03aad057f361102d3a7e0ae638462
+f31a2a0523e01feda8f41cc22512ed70ee59b3ad