Sub-orchestrators now available (#157)

Author: davidmrdavid

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -9,7 +9,8 @@
 from ..models.Task import Task
 from ..models.TokenSource import TokenSource
 from ..tasks import call_activity_task, task_all, task_any, call_activity_with_retry_task, \
-    wait_for_external_event_task, continue_as_new, new_uuid, call_http, create_timer_task
+    wait_for_external_event_task, continue_as_new, new_uuid, call_http, create_timer_task, \
+    call_sub_orchestrator_task, call_sub_orchestrator_with_retry_task
 from azure.functions._durable_functions import _deserialize_custom_object
 
 
@@ -30,6 +31,7 @@ def __init__(self,
         self._parent_instance_id: str = parentInstanceId
         self._custom_status: Any = None
         self._new_uuid_counter: int = 0
+        self._sub_orchestrator_counter: int = 0
         self.call_activity = lambda n, i=None: call_activity_task(
             state=self.histories,
             name=n,
@@ -40,6 +42,21 @@ def __init__(self,
                 retry_options=o,
                 name=n,
                 input_=i)
+        self.call_sub_orchestrator = \
+            lambda n, i=None, _id=None: call_sub_orchestrator_task(
+                context=self,
+                state=self.histories,
+                name=n,
+                input_=i,
+                instance_id=_id)
+        self.call_sub_orchestrator_with_retry = \
+            lambda n, o, i=None, _id=None: call_sub_orchestrator_with_retry_task(
+                context=self,
+                state=self.histories,
+                retry_options=o,
+                name=n,
+                input_=i,
+                instance_id=_id)
         self.call_http = lambda method, uri, content=None, headers=None, token_source=None: \
             call_http(
                 state=self.histories, method=method, uri=uri, content=content, headers=headers,

azure/durable_functions/models/actions/CallSubOrchestratorAction.py

@@ -0,0 +1,39 @@
+from typing import Any, Dict, Optional, Union
+
+from .Action import Action
+from .ActionType import ActionType
+from ..utils.json_utils import add_attrib
+from json import dumps
+from azure.functions._durable_functions import _serialize_custom_object
+
+
+class CallSubOrchestratorAction(Action):
+    """Defines the structure of the Call SubOrchestrator object."""
+
+    def __init__(self, function_name: str, _input: Optional[Any] = None, instance_id: str = ""):
+        self.function_name: str = function_name
+        self._input: str = dumps(_input, default=_serialize_custom_object)
+        self.instance_id: str = instance_id
+
+        if not self.function_name:
+            raise ValueError("function_name cannot be empty")
+
+    @property
+    def action_type(self) -> int:
+        """Get the type of action this class represents."""
+        return ActionType.CALL_SUB_ORCHESTRATOR
+
+    def to_json(self) -> Dict[str, Union[str, int]]:
+        """Convert object into a json dictionary.
+
+        Returns
+        -------
+        Dict[str, Union(str, int)]
+            The instance of the class converted into a json dictionary
+        """
+        json_dict: Dict[str, Union[str, int]] = {}
+        add_attrib(json_dict, self, 'action_type', 'actionType')
+        add_attrib(json_dict, self, 'function_name', 'functionName')
+        add_attrib(json_dict, self, '_input', 'input')
+        add_attrib(json_dict, self, 'instance_id', 'instance_id')
+        return json_dict

azure/durable_functions/models/actions/CallSubOrchestratorWithRetryAction.py

@@ -0,0 +1,44 @@
+from typing import Any, Dict, Union, Optional
+
+from .Action import Action
+from .ActionType import ActionType
+from ..utils.json_utils import add_attrib, add_json_attrib
+from json import dumps
+from ..RetryOptions import RetryOptions
+from azure.functions._durable_functions import _serialize_custom_object
+
+
+class CallSubOrchestratorWithRetryAction(Action):
+    """Defines the structure of the Call SubOrchestrator object."""
+
+    def __init__(self, function_name: str, retry_options: RetryOptions,
+                 _input: Optional[Any] = None,
+                 instance_id: str = ""):
+        self.function_name: str = function_name
+        self._input: str = dumps(_input, default=_serialize_custom_object)
+        self.retry_options: RetryOptions = retry_options
+        self.instance_id: str = instance_id
+
+        if not self.function_name:
+            raise ValueError("function_name cannot be empty")
+
+    @property
+    def action_type(self) -> int:
+        """Get the type of action this class represents."""
+        return ActionType.CALL_SUB_ORCHESTRATOR_WITH_RETRY
+
+    def to_json(self) -> Dict[str, Union[str, int]]:
+        """Convert object into a json dictionary.
+
+        Returns
+        -------
+        Dict[str, Union(str, int)]
+            The instance of the class converted into a json dictionary
+        """
+        json_dict: Dict[str, Union[str, int]] = {}
+        add_attrib(json_dict, self, 'action_type', 'actionType')
+        add_attrib(json_dict, self, 'function_name', 'functionName')
+        add_attrib(json_dict, self, '_input', 'input')
+        add_json_attrib(json_dict, self, 'retry_options', 'retryOptions')
+        add_attrib(json_dict, self, 'instance_id', 'instance_id')
+        return json_dict

azure/durable_functions/models/actions/__init__.py

@@ -3,6 +3,7 @@
 from .ActionType import ActionType
 from .CallActivityAction import CallActivityAction
 from .CallActivityWithRetryAction import CallActivityWithRetryAction
+from .CallSubOrchestratorAction import CallSubOrchestratorAction
 from .WaitForExternalEventAction import WaitForExternalEventAction
 from .CallHttpAction import CallHttpAction
 from .CreateTimerAction import CreateTimerAction
@@ -12,6 +13,7 @@
     'ActionType',
     'CallActivityAction',
     'CallActivityWithRetryAction',
+    'CallSubOrchestratorAction',
     'CallHttpAction',
     'WaitForExternalEventAction',
     'CreateTimerAction'

azure/durable_functions/tasks/__init__.py

@@ -1,6 +1,8 @@
 """Contains the definitions for the functions that enable scheduling of activities."""
 from .call_activity import call_activity_task
 from .call_activity_with_retry import call_activity_with_retry_task
+from .call_suborchestrator import call_sub_orchestrator_task
+from .call_suborchestrator_with_retry import call_sub_orchestrator_with_retry_task
 from .task_all import task_all
 from .task_any import task_any
 from .task_utilities import should_suspend
@@ -13,6 +15,8 @@
 __all__ = [
     'call_activity_task',
     'call_activity_with_retry_task',
+    'call_sub_orchestrator_task',
+    'call_sub_orchestrator_with_retry_task',
     'call_http',
     'continue_as_new',
     'new_uuid',

azure/durable_functions/tasks/call_suborchestrator.py

@@ -0,0 +1,67 @@
+from typing import List, Any, Optional
+
+from ..models.Task import (
+    Task)
+from ..models.actions.CallSubOrchestratorAction import CallSubOrchestratorAction
+from ..models.history import HistoryEvent
+from .task_utilities import set_processed, parse_history_event, \
+    find_sub_orchestration_created, find_sub_orchestration_completed, \
+    find_sub_orchestration_failed
+
+
+def call_sub_orchestrator_task(
+        context,
+        state: List[HistoryEvent],
+        name: str,
+        input_: Optional[Any] = None,
+        instance_id: str = "") -> Task:
+    """Determine the state of Scheduling a sub-orchestrator for execution.
+
+    Parameters
+    ----------
+    context: 'DurableOrchestrationContext':
+        A reference to the orchestration context.
+    state: List[HistoryEvent]
+        The list of history events to search to determine the current state of the activity.
+    name: str
+        The name of the activity function to schedule.
+    input_: Optional[Any]
+        The JSON-serializable input to pass to the activity function. Defaults to None.
+    instance_id: str
+        The instance ID of the sub-orchestrator to call. Defaults to "".
+
+    Returns
+    -------
+    Task
+        A Durable Task that completes when the called sub-orchestrator completes or fails.
+    """
+    new_action = CallSubOrchestratorAction(name, input_, instance_id)
+
+    task_scheduled = find_sub_orchestration_created(
+        state, name, context=context, instance_id=instance_id)
+    task_completed = find_sub_orchestration_completed(state, task_scheduled)
+    task_failed = find_sub_orchestration_failed(state, task_scheduled)
+    set_processed([task_scheduled, task_completed, task_failed])
+
+    if task_completed is not None:
+        return Task(
+            is_completed=True,
+            is_faulted=False,
+            action=new_action,
+            result=parse_history_event(task_completed),
+            timestamp=task_completed.timestamp,
+            id_=task_completed.TaskScheduledId)
+
+    if task_failed is not None:
+        return Task(
+            is_completed=True,
+            is_faulted=True,
+            action=new_action,
+            result=task_failed.Reason,
+            timestamp=task_failed.timestamp,
+            id_=task_failed.TaskScheduledId,
+            exc=Exception(
+                f"{task_failed.Reason} \n {task_failed.Details}")
+        )
+
+    return Task(is_completed=False, is_faulted=False, action=new_action)

azure/durable_functions/tasks/call_suborchestrator_with_retry.py

@@ -0,0 +1,79 @@
+from typing import List, Any, Optional
+
+from ..models.Task import (
+    Task)
+from ..models.actions.CallSubOrchestratorWithRetryAction import CallSubOrchestratorWithRetryAction
+from ..models.RetryOptions import RetryOptions
+from ..models.history import HistoryEvent
+from .task_utilities import set_processed, parse_history_event, \
+    find_sub_orchestration_created, find_sub_orchestration_completed, \
+    find_sub_orchestration_failed, find_task_retry_timer_fired, find_task_retry_timer_created
+
+
+def call_sub_orchestrator_with_retry_task(
+        context,
+        state: List[HistoryEvent],
+        retry_options: RetryOptions,
+        name: str,
+        input_: Optional[Any] = None,
+        instance_id: str = "") -> Task:
+    """Determine the state of Scheduling a sub-orchestrator for execution, with retry options.
+
+    Parameters
+    ----------
+    context: 'DurableOrchestrationContext':
+        A reference to the orchestration context.
+    state: List[HistoryEvent]
+        The list of history events to search to determine the current state of the activity.
+    retry_options: RetryOptions
+        The settings for retrying this sub-orchestrator in case of a failure.
+    name: str
+        The name of the activity function to schedule.
+    input_: Optional[Any]
+        The JSON-serializable input to pass to the activity function. Defaults to None.
+    instance_id: str
+        The instance ID of the sub-orchestrator to call. Defaults to "".
+
+    Returns
+    -------
+    Task
+        A Durable Task that completes when the called sub-orchestrator completes or fails.
+    """
+    new_action = CallSubOrchestratorWithRetryAction(name, retry_options, input_, instance_id)
+    for attempt in range(retry_options.max_number_of_attempts):
+        task_scheduled = find_sub_orchestration_created(
+            state, name, context=context, instance_id=instance_id)
+        task_completed = find_sub_orchestration_completed(state, task_scheduled)
+        task_failed = find_sub_orchestration_failed(state, task_scheduled)
+        task_retry_timer = find_task_retry_timer_created(state, task_failed)
+        task_retry_timer_fired = find_task_retry_timer_fired(
+            state, task_retry_timer)
+        set_processed([task_scheduled, task_completed,
+                       task_failed, task_retry_timer, task_retry_timer_fired])
+
+        if not task_scheduled:
+            break
+
+        if task_completed is not None:
+            return Task(
+                is_completed=True,
+                is_faulted=False,
+                action=new_action,
+                result=parse_history_event(task_completed),
+                timestamp=task_completed.timestamp,
+                id_=task_completed.TaskScheduledId)
+
+        if task_failed and task_retry_timer and attempt + 1 >= \
+                retry_options.max_number_of_attempts:
+            return Task(
+                is_completed=True,
+                is_faulted=True,
+                action=new_action,
+                result=task_failed.Reason,
+                timestamp=task_failed.timestamp,
+                id_=task_failed.TaskScheduledId,
+                exc=Exception(
+                    f"{task_failed.Reason} \n {task_failed.Details}")
+            )
+
+    return Task(is_completed=False, is_faulted=False, action=new_action)