Add orchestration trigger for durable functions (#48)

* Add orchestration & activity trigger for durable function

Update durable triggers

* Add dunder str method on OrchestrationContext

* Added unittest cases

* Fix activity trigger when there's no content

Author: Hazhzeng

azure/functions/__init__.py

@@ -7,6 +7,7 @@
 from ._http_wsgi import WsgiMiddleware # NoQA
 from ._queue import QueueMessage  # NoQA
 from ._servicebus import ServiceBusMessage  # NoQA
+from ._durable_functions import OrchestrationContext  # NoQA
 from .meta import get_binding_registry  # NoQA
 
 # Import binding implementations to register them
@@ -18,6 +19,7 @@
 from . import queue  # NoQA
 from . import servicebus  # NoQA
 from . import timer  # NoQA
+from . import durable_functions  # NoQA
 
 
 __all__ = (
@@ -35,9 +37,12 @@
     'EventHubEvent',
     'HttpRequest',
     'HttpResponse',
-    'WsgiMiddleware',
     'InputStream',
+    'OrchestrationContext',
     'QueueMessage',
     'ServiceBusMessage',
     'TimerRequest',
+
+    # Middlewares
+    'WsgiMiddleware',
 )

azure/functions/_abc.py

@@ -300,3 +300,10 @@ def enqueued_time(self) -> typing.Optional[datetime.datetime]:
     @abc.abstractmethod
     def offset(self) -> typing.Optional[str]:
         pass
+
+
+class OrchestrationContext(abc.ABC):
+    @property
+    @abc.abstractmethod
+    def body(self) -> str:
+        pass

azure/functions/_durable_functions.py

@@ -0,0 +1,30 @@
+from typing import Union
+from . import _abc
+
+
+class OrchestrationContext(_abc.OrchestrationContext):
+    """A durable function orchestration context.
+
+    :param str body:
+        The body of orchestration context json.
+    """
+
+    def __init__(self,
+                 body: Union[str, bytes]) -> None:
+        if isinstance(body, str):
+            self.__body = body
+        if isinstance(body, bytes):
+            self.__body = body.decode('utf-8')
+
+    @property
+    def body(self) -> str:
+        return self.__body
+
+    def __repr__(self):
+        return (
+            f'<azure.OrchestrationContext '
+            f'body={self.body}>'
+        )
+
+    def __str__(self):
+        return self.__body

azure/functions/durable_functions.py

@@ -0,0 +1,46 @@
+from typing import Any
+from azure.functions import _durable_functions
+
+from . import meta
+
+
+# Durable Function Orchestration Trigger
+class OrchestrationTriggerConverter(meta.InConverter,
+                                    binding='orchestrationTrigger',
+                                    trigger=True):
+    @classmethod
+    def check_input_type_annotation(cls, pytype):
+        return issubclass(pytype, _durable_functions.OrchestrationContext)
+
+    @classmethod
+    def decode(cls,
+               data: meta.Datum, *,
+               trigger_metadata) -> _durable_functions.OrchestrationContext:
+        return _durable_functions.OrchestrationContext(data.value)
+
+    @classmethod
+    def has_implicit_output(cls) -> bool:
+        return True
+
+
+# Durable Function Activity Trigger
+class ActivityTriggerConverter(meta.InConverter,
+                               binding='activityTrigger',
+                               trigger=True):
+    @classmethod
+    def check_input_type_annotation(cls, pytype):
+        # Activity Trigger's arguments should accept any types
+        return True
+
+    @classmethod
+    def decode(cls,
+               data: meta.Datum, *,
+               trigger_metadata) -> Any:
+        if getattr(data, 'value', None) is not None:
+            return data.value
+
+        return data
+
+    @classmethod
+    def has_implicit_output(cls) -> bool:
+        return True

azure/functions/meta.py

@@ -266,6 +266,10 @@ def check_input_type_annotation(cls, pytype: type) -> bool:
     def decode(cls, data: Datum, *, trigger_metadata) -> typing.Any:
         raise NotImplementedError
 
+    @abc.abstractclassmethod
+    def has_implicit_output(cls) -> bool:
+        return False
+
 
 class OutConverter(_BaseConverter, binding=None):
 

tests/test_durable_functions.py

@@ -0,0 +1,95 @@
+import unittest
+import json
+
+from azure.functions.durable_functions import (
+    OrchestrationTriggerConverter,
+    ActivityTriggerConverter
+)
+from azure.functions._durable_functions import OrchestrationContext
+from azure.functions.meta import Datum
+
+
+class TestDurableFunctions(unittest.TestCase):
+    def test_orchestration_context_string_body(self):
+        raw_string = '{ "name": "great function" }'
+        context = OrchestrationContext(raw_string)
+        self.assertIsNotNone(getattr(context, 'body', None))
+
+        content = json.loads(context.body)
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_context_string_cast(self):
+        raw_string = '{ "name": "great function" }'
+        context = OrchestrationContext(raw_string)
+        self.assertEqual(str(context), raw_string)
+
+        content = json.loads(str(context))
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_context_bytes_body(self):
+        raw_bytes = '{ "name": "great function" }'.encode('utf-8')
+        context = OrchestrationContext(raw_bytes)
+        self.assertIsNotNone(getattr(context, 'body', None))
+
+        content = json.loads(context.body)
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_context_bytes_cast(self):
+        raw_bytes = '{ "name": "great function" }'.encode('utf-8')
+        context = OrchestrationContext(raw_bytes)
+        self.assertIsNotNone(getattr(context, 'body', None))
+
+        content = json.loads(context.body)
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_trigger_converter(self):
+        datum = Datum(value='{ "name": "great function" }',
+                      type=str)
+        otc = OrchestrationTriggerConverter.decode(datum,
+                                                   trigger_metadata=None)
+        content = json.loads(otc.body)
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_trigger_converter_type(self):
+        datum = Datum(value='{ "name": "great function" }'.encode('utf-8'),
+                      type=bytes)
+        otc = OrchestrationTriggerConverter.decode(datum,
+                                                   trigger_metadata=None)
+        content = json.loads(otc.body)
+        self.assertEqual(content.get('name'), 'great function')
+
+    def test_orchestration_trigger_check_good_annotation(self):
+        for dt in (OrchestrationContext,):
+            self.assertTrue(
+                OrchestrationTriggerConverter.check_input_type_annotation(dt)
+            )
+
+    def test_orchestration_trigger_check_bad_annotation(self):
+        for dt in (str, bytes, int):
+            self.assertFalse(
+                OrchestrationTriggerConverter.check_input_type_annotation(dt)
+            )
+
+    def test_orchestration_trigger_has_implicit_return(self):
+        self.assertTrue(
+            OrchestrationTriggerConverter.has_implicit_output()
+        )
+
+    def test_activity_trigger_accepts_any_types(self):
+        datum_set = {
+            Datum('string', str),
+            Datum(123, int),
+            Datum(1234.56, float),
+            Datum('string'.encode('utf-8'), bytes),
+            Datum(Datum('{ "json": true }', str), Datum)
+        }
+
+        for datum in datum_set:
+            out = ActivityTriggerConverter.decode(datum, trigger_metadata=None)
+            self.assertEqual(out, datum.value)
+            self.assertEqual(type(out), datum.type)
+
+    def test_activity_trigger_has_implicit_return(self):
+        self.assertTrue(
+            ActivityTriggerConverter.has_implicit_output()
+        )

tests/test_http.py

@@ -70,3 +70,11 @@ def test_http_output_type(self):
         )
         self.assertTrue(check_output_type(func.HttpResponse))
         self.assertTrue(check_output_type(str))
+
+    def test_http_request_should_not_have_implicit_output(self):
+        self.assertFalse(http.HttpRequestConverter.has_implicit_output())
+
+    def test_http_response_does_not_have_explicit_output(self):
+        self.assertIsNone(
+            getattr(http.HttpResponseConverter, 'has_implicit_output', None)
+        )