Promote dev to master (#125)

Author: davidmrdavid

azure-pipelines.yml

@@ -8,7 +8,9 @@ trigger:
     include:
     - master
     - dev
-
+  tags:
+    include:
+    - v*
 
 variables:     
   agentPool: 'ubuntu-latest' # 'Ubuntu-16.04'

azure/durable_functions/grpc/protobuf/DurableRpc.proto

@@ -13,6 +13,7 @@
 from .OrchestrationRuntimeStatus import OrchestrationRuntimeStatus
 from ..models import DurableOrchestrationBindings
 from .utils.http_utils import get_async_request, post_async_request, delete_async_request
+from azure.functions._durable_functions import _serialize_custom_object
 
 
 class DurableOrchestrationClient:
@@ -166,7 +167,7 @@ async def raise_event(self, instance_id, event_name, event_data=None,
 
         switch_statement = {
             202: lambda: None,
-            410: lambda: None,
+            410: lambda: f"Instance with ID {instance_id} is gone: either completed or failed",
             404: lambda: f"No instance with ID {instance_id} found.",
             400: lambda: "Only application/json request content is supported"
         }
@@ -432,8 +433,27 @@ def _create_http_response(status_code: int, body: Any) -> func.HttpResponse:
         return func.HttpResponse(**response_args)
 
     @staticmethod
-    def _get_json_input(client_input: object) -> object:
-        return json.dumps(client_input) if client_input is not None else None
+    def _get_json_input(client_input: object) -> str:
+        """Serialize the orchestrator input.
+
+        Parameters
+        ----------
+        client_input: object
+            The client's input, which we need to serialize
+
+        Returns
+        -------
+        str
+            A string representing the JSON-serialization of `client_input`
+
+        Exceptions
+        ----------
+        TypeError
+            If the JSON serialization failed, see `serialize_custom_object`
+        """
+        if client_input is not None:
+            return json.dumps(client_input, default=_serialize_custom_object)
+        return None
 
     @staticmethod
     def _replace_url_origin(request_url, value_url):

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -10,6 +10,7 @@
 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
+from azure.functions._durable_functions import _deserialize_custom_object
 
 
 class DurableOrchestrationContext:
@@ -27,6 +28,7 @@ def __init__(self,
         self._instance_id: str = instanceId
         self._is_replaying: bool = isReplaying
         self._parent_instance_id: str = parentInstanceId
+        self._custom_status: Any = None
         self._new_uuid_counter: int = 0
         self.call_activity = lambda n, i=None: call_activity_task(
             state=self.histories,
@@ -78,6 +80,8 @@ def from_json(cls, json_string: str):
         DurableOrchestrationContext
             New instance of the durable orchestration context class
         """
+        # We should consider parsing the `Input` field here as well,
+        # intead of doing so lazily when `get_input` is called.
         json_dict = json.loads(json_string)
         return cls(**json_dict)
 
@@ -164,7 +168,8 @@ def call_sub_orchestrator(self,
 
     def get_input(self) -> str:
         """Get the orchestration input."""
-        return self._input
+        return None if self._input is None else json.loads(self._input,
+                                                           object_hook=_deserialize_custom_object)
 
     def new_uuid(self) -> str:
         """Create a new UUID that is safe for replay within an orchestration or operation.
@@ -221,6 +226,23 @@ def task_any(self, activities: List[Task]) -> TaskSet:
         """
         raise NotImplementedError("This is a placeholder.")
 
+    def set_custom_status(self, status: Any):
+        """Set the customized orchestration status for your orchestrator function.
+
+        This status is also returned by the orchestration client through the get_status API
+
+        Parameters
+        ----------
+        status : str
+            Customized status provided by the orchestrator
+        """
+        self._custom_status = status
+
+    @property
+    def custom_status(self):
+        """Get customized status of current orchestration."""
+        return self._custom_status
+
     @property
     def histories(self):
         """Get running history of tasks that have been scheduled."""

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -3,6 +3,8 @@
 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 CallActivityAction(Action):
@@ -13,7 +15,8 @@ class CallActivityAction(Action):
 
     def __init__(self, function_name: str, input_=None):
         self.function_name: str = function_name
-        self.input_ = input_
+        # It appears that `.input_` needs to be JSON-serializable at this point
+        self.input_ = dumps(input_, default=_serialize_custom_object)
 
         if not self.function_name:
             raise ValueError("function_name cannot be empty")

azure/durable_functions/models/actions/CallActivityAction.py

@@ -32,7 +32,6 @@ def __init__(self,
         :param activity_func: Generator function to orchestrate.
         """
         self.fn: Callable[[DurableOrchestrationContext], Iterator[Any]] = activity_func
-        self.customStatus: Any = None
 
     def handle(self, context: DurableOrchestrationContext):
         """Handle the orchestration of the user defined generator function.
@@ -46,9 +45,23 @@ def handle(self, context: DurableOrchestrationContext):
         the durable extension.
         """
         self.durable_context = context
-
-        self.generator = self.fn(self.durable_context)
+        self.generator = None
         suspended = False
+
+        fn_output = self.fn(self.durable_context)
+        # If `fn_output` is not an Iterator, then the orchestrator
+        # function does not make use of its context parameter. If so,
+        # `fn_output` is the return value instead of a generator
+        if isinstance(fn_output, Iterator):
+            self.generator = fn_output
+
+        else:
+            orchestration_state = OrchestratorState(
+                is_done=True,
+                output=fn_output,
+                actions=self.durable_context.actions,
+                custom_status=self.durable_context.custom_status)
+            return orchestration_state.to_json_string()
         try:
             generation_state = self._generate_next(None)
 
@@ -60,7 +73,7 @@ def handle(self, context: DurableOrchestrationContext):
                         is_done=False,
                         output=None,
                         actions=self.durable_context.actions,
-                        custom_status=self.customStatus)
+                        custom_status=self.durable_context.custom_status)
                     suspended = True
                     continue
 
@@ -79,14 +92,14 @@ def handle(self, context: DurableOrchestrationContext):
                 is_done=True,
                 output=sie.value,
                 actions=self.durable_context.actions,
-                custom_status=self.customStatus)
+                custom_status=self.durable_context.custom_status)
         except Exception as e:
             orchestration_state = OrchestratorState(
                 is_done=False,
                 output=None,  # Should have no output, after generation range
                 actions=self.durable_context.actions,
                 error=str(e),
-                custom_status=self.customStatus)
+                custom_status=self.durable_context.custom_status)
 
         return orchestration_state.to_json_string()
 

azure/durable_functions/orchestrator.py

@@ -1,5 +1,6 @@
 import json
 from ..models.history import HistoryEventType
+from azure.functions._durable_functions import _deserialize_custom_object
 
 
 def should_suspend(partial_result) -> bool:
@@ -15,12 +16,14 @@ def parse_history_event(directive_result):
     if event_type is None:
         raise ValueError("EventType is not found in task object")
 
+    # We provide the ability to deserialize custom objects, because the output of this
+    # will be passed directly to the orchestrator as the output of some activity
     if event_type == HistoryEventType.EVENT_RAISED:
-        return json.loads(directive_result.Input)
+        return json.loads(directive_result.Input, object_hook=_deserialize_custom_object)
     if event_type == HistoryEventType.SUB_ORCHESTRATION_INSTANCE_CREATED:
-        return json.loads(directive_result.Result)
+        return json.loads(directive_result.Result, object_hook=_deserialize_custom_object)
     if event_type == HistoryEventType.TASK_COMPLETED:
-        return json.loads(directive_result.Result)
+        return json.loads(directive_result.Result, object_hook=_deserialize_custom_object)
     return None
 
 

azure/durable_functions/tasks/task_utilities.py

@@ -5,10 +5,11 @@ def tests(session):
     # same as pip install -r -requirements.txt
     session.install("-r", "requirements.txt")
     session.install("pytest")
-    session.run("pytest","-v","tests")
+    session.run("pytest", "-v", "tests")
+
 
 @nox.session(python="3.7")
 def lint(session):
     session.install("flake8")
-    session.run("flake8","./azure/**py")
-
+    session.install("flake8-docstrings")
+    session.run("flake8", "./azure/")

noxfile.py

@@ -0,0 +1,5 @@
+.git*
+.vscode
+local.settings.json
+test
+.venv