Merge branch 'main' into filinto/asyncio

Author: filintod

CODEOWNERS

@@ -0,0 +1,2 @@
+# These owners are the maintainers and approvers of this repo
+*       @dapr/maintainers-python-sdk @dapr/approvers-python-sdk

durabletask/internal/helpers.py

@@ -239,14 +239,15 @@ def new_create_timer_action(id: int, fire_at: datetime) -> pb.OrchestratorAction
     )
 
 
-def new_schedule_task_action(
-    id: int, name: str, encoded_input: Optional[str]
-) -> pb.OrchestratorAction:
+def new_schedule_task_action(id: int, name: str, encoded_input: Optional[str], router: Optional[pb.TaskRouter] = None) -> pb.OrchestratorAction:
     return pb.OrchestratorAction(
         id=id,
         scheduleTask=pb.ScheduleTaskAction(
-            name=name, input=get_string_value(encoded_input)
+            name=name,
+            input=get_string_value(encoded_input),
+            router=router,
         ),
+        router=router,
     )
 
 
@@ -257,13 +258,20 @@ def new_timestamp(dt: datetime) -> timestamp_pb2.Timestamp:
 
 
 def new_create_sub_orchestration_action(
-    id: int, name: str, instance_id: Optional[str], encoded_input: Optional[str]
-) -> pb.OrchestratorAction:
+        id: int,
+        name: str,
+        instance_id: Optional[str],
+        encoded_input: Optional[str],
+        router: Optional[pb.TaskRouter] = None) -> pb.OrchestratorAction:
     return pb.OrchestratorAction(
         id=id,
         createSubOrchestration=pb.CreateSubOrchestrationAction(
-            name=name, instanceId=instance_id, input=get_string_value(encoded_input)
+            name=name,
+            instanceId=instance_id,
+            input=get_string_value(encoded_input),
+            router=router,
         ),
+        router=router,
     )
 
 

durabletask/task.py

@@ -146,13 +146,10 @@ def create_timer(self, fire_at: Union[datetime, timedelta]) -> "Task[Any]":
         pass
 
     @abstractmethod
-    def call_activity(
-        self,
-        activity: Union[Activity[TInput, TOutput], str],
-        *,
-        input: Optional[TInput] = None,
-        retry_policy: Optional[RetryPolicy] = None,
-    ) -> Task[TOutput]:
+    def call_activity(self, activity: Union[Activity[TInput, TOutput], str], *,
+                      input: Optional[TInput] = None,
+                      retry_policy: Optional[RetryPolicy] = None,
+                      app_id: Optional[str] = None) -> Task[TOutput]:
         """Schedule an activity for execution.
 
         Parameters
@@ -163,6 +160,8 @@ def call_activity(
             The JSON-serializable input (or None) to pass to the activity.
         retry_policy: Optional[RetryPolicy]
             The retry policy to use for this activity call.
+        app_id: Optional[str]
+            The app ID that will execute the activity. If not specified, the activity will be executed by the same app as the orchestrator.
 
         Returns
         -------
@@ -172,14 +171,11 @@ def call_activity(
         pass
 
     @abstractmethod
-    def call_sub_orchestrator(
-        self,
-        orchestrator: Orchestrator[TInput, TOutput],
-        *,
-        input: Optional[TInput] = None,
-        instance_id: Optional[str] = None,
-        retry_policy: Optional[RetryPolicy] = None,
-    ) -> Task[TOutput]:
+    def call_sub_orchestrator(self, orchestrator: Orchestrator[TInput, TOutput], *,
+                              input: Optional[TInput] = None,
+                              instance_id: Optional[str] = None,
+                              retry_policy: Optional[RetryPolicy] = None,
+                              app_id: Optional[str] = None) -> Task[TOutput]:
         """Schedule sub-orchestrator function for execution.
 
         Parameters
@@ -193,6 +189,8 @@ def call_sub_orchestrator(
             random UUID will be used.
         retry_policy: Optional[RetryPolicy]
             The retry policy to use for this sub-orchestrator call.
+        app_id: Optional[str]
+            The app ID that will execute the sub-orchestrator. If not specified, the sub-orchestrator will be executed by the same app as the orchestrator.
 
         Returns
         -------

durabletask/worker.py

@@ -818,6 +818,7 @@ def __init__(self, instance_id: str):
         self._sequence_number = 0
         self._current_utc_datetime = datetime(1000, 1, 1)
         self._instance_id = instance_id
+        self._app_id = None
         self._completion_status: Optional[pb.OrchestrationStatus] = None
         self._received_events: dict[str, list[Any]] = {}
         self._pending_events: dict[str, list[task.CompletableTask]] = {}
@@ -970,6 +971,10 @@ def next_sequence_number(self) -> int:
         self._sequence_number += 1
         return self._sequence_number
 
+    @property
+    def app_id(self) -> str:
+        return self._app_id
+
     @property
     def instance_id(self) -> str:
         return self._instance_id
@@ -1042,55 +1047,65 @@ def create_timer_internal(
         return timer_task
 
     def call_activity(
-        self,
-        activity: Union[task.Activity[TInput, TOutput], str],
-        *,
-        input: Optional[TInput] = None,
-        retry_policy: Optional[task.RetryPolicy] = None,
+            self,
+            activity: Union[task.Activity[TInput, TOutput], str],
+            *,
+            input: Optional[TInput] = None,
+            retry_policy: Optional[task.RetryPolicy] = None,
+            app_id: Optional[str] = None,
     ) -> task.Task[TOutput]:
         id = self.next_sequence_number()
 
         self.call_activity_function_helper(
-            id, activity, input=input, retry_policy=retry_policy, is_sub_orch=False
+            id, activity, input=input, retry_policy=retry_policy, is_sub_orch=False, app_id=app_id
         )
         return self._pending_tasks.get(id, task.CompletableTask())
 
     def call_sub_orchestrator(
-        self,
-        orchestrator: Union[task.Orchestrator[TInput, TOutput], str],
-        *,
-        input: Optional[TInput] = None,
-        instance_id: Optional[str] = None,
-        retry_policy: Optional[task.RetryPolicy] = None,
+            self,
+            orchestrator: Union[task.Orchestrator[TInput, TOutput], str],
+            *,
+            input: Optional[TInput] = None,
+            instance_id: Optional[str] = None,
+            retry_policy: Optional[task.RetryPolicy] = None,
+            app_id: Optional[str] = None,
     ) -> task.Task[TOutput]:
         id = self.next_sequence_number()
-        orchestrator_name = (
-            orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
-        )
+        if isinstance(orchestrator, str):
+            orchestrator_name = orchestrator
+        else:
+            orchestrator_name = task.get_name(orchestrator)
         self.call_activity_function_helper(
             id,
             orchestrator_name,
             input=input,
             retry_policy=retry_policy,
             is_sub_orch=True,
             instance_id=instance_id,
+            app_id=app_id,
         )
         return self._pending_tasks.get(id, task.CompletableTask())
 
     def call_activity_function_helper(
-        self,
-        id: Optional[int],
-        activity_function: Union[task.Activity[TInput, TOutput], str],
-        *,
-        input: Optional[TInput] = None,
-        retry_policy: Optional[task.RetryPolicy] = None,
-        is_sub_orch: bool = False,
-        instance_id: Optional[str] = None,
-        fn_task: Optional[task.CompletableTask[TOutput]] = None,
+            self,
+            id: Optional[int],
+            activity_function: Union[task.Activity[TInput, TOutput], str],
+            *,
+            input: Optional[TInput] = None,
+            retry_policy: Optional[task.RetryPolicy] = None,
+            is_sub_orch: bool = False,
+            instance_id: Optional[str] = None,
+            fn_task: Optional[task.CompletableTask[TOutput]] = None,
+            app_id: Optional[str] = None,
     ):
         if id is None:
             id = self.next_sequence_number()
 
+        router = pb.TaskRouter()
+        router.sourceAppID = self._app_id
+        if app_id is not None:
+            router.targetAppID = app_id
+
         if fn_task is None:
             encoded_input = shared.to_json(input) if input is not None else None
         else:
@@ -1132,15 +1147,15 @@ def call_activity_function_helper(
                 if isinstance(activity_function, str)
                 else task.get_name(activity_function)
             )
-            action = ph.new_schedule_task_action(id, name, encoded_input)
+            action = ph.new_schedule_task_action(id, name, encoded_input, router)
         else:
             if instance_id is None:
                 # Create a deteministic instance ID based on the parent instance ID
                 instance_id = f"{self.instance_id}:{id:04x}"
             if not isinstance(activity_function, str):
                 raise ValueError("Orchestrator function name must be a string")
             action = ph.new_create_sub_orchestration_action(
-                id, activity_function, instance_id, encoded_input
+                id, activity_function, instance_id, encoded_input, router
             )
         self._pending_actions[id] = action
 
@@ -1285,6 +1300,11 @@ def process_event(
             if event.HasField("orchestratorStarted"):
                 ctx.current_utc_datetime = event.timestamp.ToDatetime()
             elif event.HasField("executionStarted"):
+                if event.router.targetAppID:
+                    ctx._app_id = event.router.targetAppID
+                else:
+                    ctx._app_id = event.router.sourceAppID
+
                 # TODO: Check if we already started the orchestration
                 fn = self._registry.get_orchestrator(event.executionStarted.name)
                 if fn is None:
@@ -1398,6 +1418,11 @@ def process_event(
                     else:
                         cur_task = activity_action.createSubOrchestration
                         instance_id = cur_task.instanceId
+                    if cur_task.router and cur_task.router.targetAppID:
+                        target_app_id = cur_task.router.targetAppID
+                    else:
+                        target_app_id = None
+
                     ctx.call_activity_function_helper(
                         id=activity_action.id,
                         activity_function=cur_task.name,
@@ -1406,6 +1431,7 @@ def process_event(
                         is_sub_orch=timer_task._retryable_parent._is_sub_orch,
                         instance_id=instance_id,
                         fn_task=timer_task._retryable_parent,
+                        app_id=target_app_id,
                     )
                 else:
                     ctx.resume()