[WIP] Semi-automatic type-serialization

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -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:
@@ -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/DurableOrchestrationContext.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:
@@ -79,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)
 
@@ -165,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.

azure/durable_functions/models/actions/CallActivityAction.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/tasks/task_utilities.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
 
 

samples/serialize_arguments/DurableActivity/__init__.py

@@ -0,0 +1,19 @@
+import logging
+
+def main(name):
+    """Activity function performing a specific step in the chain
+
+    Parameters
+    ----------
+    name : str
+        Name of the item to be hello'ed at 
+
+    Returns
+    -------
+    str
+        Returns a welcome string
+    """
+    logging.warning(f"Activity Triggered: {name}")
+    return name
+
+

samples/serialize_arguments/DurableActivity/function.json

@@ -0,0 +1,12 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "name": "name",
+      "type": "activityTrigger",
+      "direction": "in",
+      "datatype": "string"
+    }
+  ],
+  "disabled": false
+}

samples/serialize_arguments/DurableOrchestration/__init__.py

@@ -0,0 +1,29 @@
+import logging
+
+import azure.functions as func
+import azure.durable_functions as df
+from ..shared_code.MyClasses import SerializableClass # TODO: this import is highlight 'red' in VSCode, but works at runtime
+
+def orchestrator_function(context: df.DurableOrchestrationContext):
+    """This function provides the core function chaining orchestration logic
+
+    Parameters
+    ----------
+    context: DurableOrchestrationContext
+        This context has the past history and the durable orchestration API's to
+        create orchestrations
+
+    Returns
+    -------
+    int
+        The number contained in the input
+    """
+    input_: SerializableClass = context.get_input()
+    number: int = input_.show_number()
+
+
+    # throwaway, seems necessary for the orchestration not to fail
+    value = yield context.call_activity("DurableActivity", SerializableClass(24))
+    return 11
+
+main = df.Orchestrator.create(orchestrator_function)

samples/serialize_arguments/DurableOrchestration/function.json

@@ -0,0 +1,11 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "name": "context",
+      "type": "orchestrationTrigger",
+      "direction": "in"
+    }
+  ],
+  "disabled": false
+}

samples/serialize_arguments/DurableTrigger/__init__.py

@@ -0,0 +1,31 @@
+import logging
+
+from azure.durable_functions import DurableOrchestrationClient
+import azure.functions as func
+from ..shared_code.MyClasses import SerializableClass
+
+
+async def main(req: func.HttpRequest, starter: str, message):
+    """This function starts up the orchestrator from an HTTP endpoint
+
+    starter: str
+        A JSON-formatted string describing the orchestration context
+
+    message:
+        An azure functions http output binding, it enables us to establish
+        an http response.
+
+    Parameters
+    ----------
+    req: func.HttpRequest
+        An HTTP Request object, it can be used to parse URL
+        parameters.
+    """
+
+
+    function_name = req.route_params.get('functionName')
+    logging.info(starter)
+    client = DurableOrchestrationClient(starter)
+    instance_id = await client.start_new(function_name, client_input=SerializableClass(11))
+    response = client.create_check_status_response(req, instance_id)
+    message.set(response)

samples/serialize_arguments/DurableTrigger/function.json

@@ -0,0 +1,27 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "authLevel": "anonymous",
+      "name": "req",
+      "type": "httpTrigger",
+      "direction": "in",
+      "route": "orchestrators/{functionName}",
+      "methods": [
+        "post",
+        "get"
+      ]
+    },
+    {
+      "direction": "out",
+      "name": "message",
+      "type": "http"
+    },
+    {
+      "name": "starter",
+      "type": "orchestrationClient",
+      "direction": "in",
+      "datatype": "string"
+    }
+  ]
+}

samples/serialize_arguments/README.md

@@ -0,0 +1,35 @@
+# Serializing Arguments - Sample
+
+TBD
+
+## Usage Instructions
+
+### Create a `local.settings.json` file in this directory
+This file stores app settings, connection strings, and other settings used by local development tools. Learn more about it [here](https://docs.microsoft.com/en-us/azure/azure-functions/functions-run-local?tabs=windows%2Ccsharp%2Cbash#local-settings-file).
+For this sample, you will only need an `AzureWebJobsStorage` connection string, which you can obtain from the Azure portal.
+
+With you connection string, your `local.settings.json` file should look as follows, with `<your connection string>` replaced with the connection string you obtained from the Azure portal:
+
+```json
+{
+  "IsEncrypted": false,
+  "Values": {
+    "AzureWebJobsStorage": "<your connection string>",
+    "FUNCTIONS_WORKER_RUNTIME": "python"
+  }
+}
+```
+
+### Run the Sample
+To try this sample, run  `func host start` in this directory. If all the system requirements have been met, and
+after some initialization logs, you should see something like the following:
+
+```bash
+Http Functions:
+
+        DurableTrigger: [POST,GET] http://localhost:7071/api/orchestrators/{functionName}
+```
+
+This indicates that your `DurableTrigger` function can be reached via a `GET` or `POST` request to that URL. `DurableTrigger` starts the function-chaning orchestrator whose name is passed as a parameter to the URL. So, to start the orchestrator, which is named `DurableOrchestration`, make a GET request to `http://127.0.0.1:7071/api/orchestrators/DurableOrchestration`.
+
+And that's it! You should see a JSON response with five URLs to monitor the status of the orchestration. To learn more about this, please read [here](TODO)!

samples/serialize_arguments/host.json

@@ -0,0 +1,7 @@
+{
+  "version": "2.0",
+  "extensionBundle": {
+    "id": "Microsoft.Azure.Functions.ExtensionBundle",
+    "version": "[1.*, 2.0.0)"
+  }
+}

samples/serialize_arguments/shared_code/MyClasses.py

@@ -0,0 +1,63 @@
+import typing
+import json
+
+class SerializableClass(object):
+    """ Example serializable class.
+
+    For a custom class to be serializable in
+    Python Durable Functions, we require that
+    it include both `to_json` and `from_json`
+    a `@staticmethod`s for serializing to JSON
+    and back respectively. These get called
+    internally by the framework.
+    """
+    def __init__(self, number: int):
+        """ Construct the class
+        Parameters
+        ----------
+        number: int
+            A number to encapsulate
+        """
+        self.number = number
+
+    def show_number(self) -> int:
+        """" Returns the number value"""
+        return self.number
+
+    @staticmethod
+    def to_json(obj: object) -> str:
+        """ Serializes a `SerializableClass` instance
+        to a JSON string.
+
+        Parameters
+        ----------
+        obj: SerializableClass
+            The object to serialize
+
+        Returns
+        -------
+        json_str: str
+            A JSON-encoding of `obj`
+        """
+        return str(obj.number)
+
+    @staticmethod
+    def from_json(json_str: str) -> object:
+        """ De-serializes a JSON string to a
+        `SerializableClass` instance. It assumes
+        that the JSON string was generated via
+        `SerializableClass.to_json`
+
+        Parameters
+        ----------
+        json_str: str
+            The JSON-encoding of a `SerializableClass` instance
+
+        Returns
+        --------
+        obj: SerializableClass
+            A SerializableClass instance, de-serialized from `json_str`
+        """
+        number = int(json_str)
+        obj = SerializableClass(number)
+        return obj

tests/models/test_DurableOrchestrationContext.py

@@ -58,23 +58,22 @@ def test_added_function_context_args():
 
 
 def test_get_input_none(starting_context):
-    assert None == starting_context.get_input()
+    test = starting_context.get_input()
+    assert None == test
 
 
 def test_get_input_string():
     builder = ContextBuilder('test_function_context')
-    builder.input_ = 'Seattle'
+    builder.input_ = json.dumps('Seattle')
     context = DurableOrchestrationContext.from_json(builder.to_json_string())
-
     assert 'Seattle' == context.get_input()
 
 
 def test_get_input_json_str():
     builder = ContextBuilder('test_function_context')
-    builder.input_ = { 'city': 'Seattle' }
+    builder.input_ = json.dumps({ 'city': 'Seattle' })
     context = DurableOrchestrationContext.from_json(builder.to_json_string())
 
     result = context.get_input()
 
-    result_dict = json.loads(result)
-    assert 'Seattle' == result_dict['city']
+    assert 'Seattle' == result['city']

tests/models/test_OrchestrationState.py

@@ -23,5 +23,5 @@ def test_single_action_state_to_json_string():
     result = state.to_json_string()
     expected_result = ('{"isDone": false, "actions": [[{"actionType": 0, '
                        '"functionName": "MyFunction", "input": '
-                       '"AwesomeInput"}]]}')
+                       '"\\"AwesomeInput\\""}]]}')
     assert expected_result == result