Add missing type coersion, rich client, tests

Author: andystaples

azure-functions-durable/azure/durable_functions/decorators/durable_app.py

@@ -12,6 +12,7 @@
 
 from .metadata import OrchestrationTrigger, ActivityTrigger, EntityTrigger, \
     DurableClient
+from ..client import DurableFunctionsClient
 from ..worker import DurableFunctionsWorker
 from ..orchestrator import Orchestrator
 
@@ -47,14 +48,19 @@ def __init__(self,
 
     def _configure_orchestrator_callable(
             self,
-            wrap: Callable[[Callable[..., Any]], FunctionBuilder]
+            wrap: Callable[[Callable[..., Any]], FunctionBuilder],
+            input_type: Optional[type] = None
     ) -> Callable[[task.Orchestrator[Any, Any]], FunctionBuilder]:
         """Obtain decorator to construct an Orchestrator class from a user-defined Function.
 
         Parameters
         ----------
         wrap: Callable
             The next decorator to be applied.
+        input_type: Optional[type]
+            The expected type for orchestration input, forwarded from the
+            ``orchestration_trigger`` decorator so a v1-style
+            ``context.get_input()`` can decode the input to that type.
 
         Returns
         -------
@@ -65,6 +71,11 @@ def _configure_orchestrator_callable(
         def decorator(orchestrator_func: task.Orchestrator[Any, Any]) -> FunctionBuilder:
             # Construct an orchestrator based on the end-user code
 
+            if input_type is not None:
+                # Stash the decorator-declared input type so the runtime can
+                # feed it to a v1-style ``context.get_input()``.
+                orchestrator_func._df_input_type = input_type  # type: ignore[attr-defined]  # noqa: E501
+
             handle = Orchestrator.create(orchestrator_func)
 
             # invoke next decorator, with the Orchestrator as input
@@ -167,7 +178,8 @@ def decorator(func: Callable[..., Any]) -> FunctionBuilder:
         return decorator
 
     def orchestration_trigger(self, context_name: str,
-                              orchestration: Optional[str] = None
+                              orchestration: Optional[str] = None,
+                              input_type: Optional[type] = None
                               ) -> Callable[[task.Orchestrator[Any, Any]], FunctionBuilder]:
         """Register an Orchestrator Function.
 
@@ -178,8 +190,12 @@ def orchestration_trigger(self, context_name: str,
         orchestration: Optional[str]
             Name of Orchestrator Function.
             The value is None by default, in which case the name of the method is used.
+        input_type: Optional[type]
+            The expected type for the orchestration input. When set, a v1-style
+            ``context.get_input()`` decodes the input payload to this type. A
+            call-site ``expected_type`` argument on ``get_input`` takes
+            precedence over this value.
         """
-        @self._configure_orchestrator_callable
         @self._build_function
         def wrap(fb: FunctionBuilder) -> FunctionBuilder:
 
@@ -191,7 +207,7 @@ def decorator() -> FunctionBuilder:
 
             return decorator()
 
-        return wrap
+        return self._configure_orchestrator_callable(wrap, input_type=input_type)
 
     def activity_trigger(self, input_name: str,
                          activity: Optional[str] = None
@@ -271,6 +287,7 @@ def durable_client_input(self,
         @self._build_function
         def wrap(fb: FunctionBuilder) -> FunctionBuilder:
             def decorator() -> FunctionBuilder:
+                self._add_rich_client(fb, client_name, DurableFunctionsClient)
                 fb.add_binding(
                     binding=DurableClient(name=client_name,
                                           task_hub=task_hub,

azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py

@@ -10,6 +10,7 @@
 from durabletask.entities import EntityInstanceId
 from durabletask.task import OrchestrationContext, RetryPolicy, Task
 
+from ..serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER
 from .token_source import TokenSource
 
 
@@ -24,16 +25,28 @@ class DurableOrchestrationContext:
     context directly instead.
     """
 
-    def __init__(self, ctx: OrchestrationContext, orchestration_input: Any = None):
+    def __init__(self,
+                 ctx: OrchestrationContext,
+                 orchestration_input: Any = None,
+                 input_type: Optional[type] = None):
         self._ctx = ctx
         self._input = orchestration_input
+        self._input_type = input_type
         self._custom_status: Any = None
         self._will_continue_as_new = False
 
     # -- input ---------------------------------------------------------------
-    def get_input(self) -> Any:
-        """Get the orchestration input."""
-        return self._input
+    def get_input(self, expected_type: Optional[type] = None) -> Any:
+        """Get the orchestration input.
+
+        When an ``expected_type`` (or the ``input_type`` declared on the
+        ``orchestration_trigger`` decorator) is available, the already-decoded
+        input is coerced to that type; otherwise the raw value is returned.
+        """
+        resolved_type = expected_type or self._input_type
+        if resolved_type is None:
+            return self._input
+        return DEFAULT_FUNCTIONS_DATA_CONVERTER.coerce(self._input, resolved_type)
 
     # -- properties ----------------------------------------------------------
     @property
@@ -227,19 +240,20 @@ def wrap_orchestrator(fn: Callable[..., Any]) -> Callable[..., Any]:
     if accepts_two_positional_args(fn):
         return fn
 
+    input_type = getattr(fn, "_df_input_type", None)
     name = getattr(fn, "__name__", "orchestrator")
 
     if inspect.isgeneratorfunction(fn):
         def _generator_wrapper(context: OrchestrationContext, _input: Any = None) -> Any:
-            adapter = DurableOrchestrationContext(context, _input)
+            adapter = DurableOrchestrationContext(context, _input, input_type)
             generator = cast("Generator[Any, Any, Any]", fn(adapter))
             result: Any = yield from generator
             return result
         _generator_wrapper.__name__ = name
         return _generator_wrapper
 
     def _wrapper(context: OrchestrationContext, _input: Any = None) -> Any:
-        adapter = DurableOrchestrationContext(context, _input)
+        adapter = DurableOrchestrationContext(context, _input, input_type)
         return fn(adapter)
     _wrapper.__name__ = name
     return _wrapper

tests/azure-functions-durable/test_decorator_compat.py

@@ -0,0 +1,148 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+
+import azure.durable_functions as df
+from azure.durable_functions import DurableFunctionsClient
+from azure.durable_functions.constants import (
+    ACTIVITY_TRIGGER,
+    DURABLE_CLIENT,
+    ENTITY_TRIGGER,
+    ORCHESTRATION_TRIGGER,
+)
+
+
+_CLIENT_CONFIG = json.dumps({
+    "taskHubName": "TestHub",
+    "requiredQueryStringParameters": "code=xyz",
+    "baseUrl": "http://localhost:7071/runtime/webhooks/durabletask",
+    "rpcBaseUrl": "http://localhost:8080/",
+})
+
+
+def _trigger(fb):
+    return fb._function.get_trigger()
+
+
+# ---------------------------------------------------------------------------
+# orchestration_trigger
+# ---------------------------------------------------------------------------
+
+def test_orchestration_trigger_v1_signature():
+    app = df.DFApp()
+
+    def my_orchestrator(context):
+        return 1
+
+    fb = app.orchestration_trigger(
+        context_name="context", orchestration="MyOrchestrator")(my_orchestrator)
+    trigger = _trigger(fb)
+    assert trigger.get_binding_name() == ORCHESTRATION_TRIGGER
+    assert trigger.name == "context"
+    assert trigger.orchestration == "MyOrchestrator"
+
+
+def test_orchestration_trigger_accepts_input_type():
+    app = df.DFApp()
+
+    def my_orchestrator(context):
+        return 1
+
+    # v1 parity: the input_type keyword must be accepted and stashed.
+    fb = app.orchestration_trigger(
+        context_name="context", input_type=dict)(my_orchestrator)
+    assert fb is not None
+    assert my_orchestrator._df_input_type is dict
+
+
+# ---------------------------------------------------------------------------
+# activity_trigger
+# ---------------------------------------------------------------------------
+
+def test_activity_trigger_v1_signature():
+    app = df.DFApp()
+
+    def my_activity(myinput):
+        return myinput
+
+    fb = app.activity_trigger(
+        input_name="myinput", activity="MyActivity")(my_activity)
+    trigger = _trigger(fb)
+    assert trigger.get_binding_name() == ACTIVITY_TRIGGER
+    assert trigger.name == "myinput"
+    assert trigger.activity == "MyActivity"
+
+
+# ---------------------------------------------------------------------------
+# entity_trigger
+# ---------------------------------------------------------------------------
+
+def test_entity_trigger_v1_signature():
+    app = df.DFApp()
+
+    def my_entity(context):
+        return None
+
+    fb = app.entity_trigger(
+        context_name="context", entity_name="MyEntity")(my_entity)
+    trigger = _trigger(fb)
+    assert trigger.get_binding_name() == ENTITY_TRIGGER
+    assert trigger.name == "context"
+    assert trigger.entity_name == "MyEntity"
+
+
+# ---------------------------------------------------------------------------
+# durable_client_input
+# ---------------------------------------------------------------------------
+
+def test_durable_client_input_v1_signature_registers_binding():
+    app = df.DFApp()
+
+    async def starter(client):
+        return None
+
+    fb = app.durable_client_input(
+        client_name="client", task_hub="hub", connection_name="conn")(starter)
+    bindings = fb._function.get_bindings()
+    client_bindings = [b for b in bindings if b.get_binding_name() == DURABLE_CLIENT]
+    assert len(client_bindings) == 1
+    binding = client_bindings[0]
+    assert binding.name == "client"
+    assert binding.task_hub == "hub"
+    assert binding.connection_name == "conn"
+
+
+async def test_durable_client_input_injects_rich_client():
+    app = df.DFApp()
+    received = {}
+
+    async def starter(client):
+        received["client"] = client
+
+    fb = app.durable_client_input(client_name="client")(starter)
+    # _add_rich_client replaces the user function with middleware that builds
+    # a DurableFunctionsClient from the binding's JSON string.
+    middleware = fb._function._func
+    await middleware(client=_CLIENT_CONFIG)
+
+    client = received["client"]
+    assert isinstance(client, DurableFunctionsClient)
+    try:
+        assert client.taskHubName == "TestHub"
+    finally:
+        await client.close()
+
+
+# ---------------------------------------------------------------------------
+# All decorators register a function builder
+# ---------------------------------------------------------------------------
+
+def test_decorators_register_function_builders():
+    app = df.DFApp()
+
+    def orch(context):
+        return 1
+
+    app.orchestration_trigger(context_name="context")(orch)
+    assert len(app._function_builders) == 1

tests/azure-functions-durable/test_serialization_compat.py

@@ -0,0 +1,71 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import pytest
+
+from azure.durable_functions.internal.serialization import (
+    DEFAULT_FUNCTIONS_DATA_CONVERTER,
+)
+
+
+class Point:
+    """Sample custom type using the v1 to_json / from_json convention."""
+
+    def __init__(self, x: int, y: int):
+        self.x = x
+        self.y = y
+
+    def to_json(self):
+        return {"x": self.x, "y": self.y}
+
+    @classmethod
+    def from_json(cls, data):
+        return cls(data["x"], data["y"])
+
+    def __eq__(self, other):
+        return isinstance(other, Point) and self.x == other.x and self.y == other.y
+
+
+def test_custom_object_round_trips():
+    point = Point(3, 4)
+    serialized = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(point)
+    assert isinstance(serialized, str)
+
+    restored = DEFAULT_FUNCTIONS_DATA_CONVERTER.deserialize(serialized)
+    assert isinstance(restored, Point)
+    assert restored == point
+
+
+def test_nested_custom_object_round_trips():
+    payload = {"points": [Point(1, 1), Point(2, 2)], "label": "path"}
+    serialized = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(payload)
+    restored = DEFAULT_FUNCTIONS_DATA_CONVERTER.deserialize(serialized)
+    assert restored["label"] == "path"
+    assert restored["points"] == [Point(1, 1), Point(2, 2)]
+
+
+@pytest.mark.parametrize("value", [
+    {"a": 1, "b": [1, 2, 3]},
+    [1, 2, 3],
+    "hello",
+    42,
+    3.14,
+    True,
+])
+def test_builtin_values_round_trip(value):
+    serialized = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(value)
+    assert isinstance(serialized, str)
+    restored = DEFAULT_FUNCTIONS_DATA_CONVERTER.deserialize(serialized)
+    assert restored == value
+
+
+def test_none_round_trips():
+    assert DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(None) is None
+    assert DEFAULT_FUNCTIONS_DATA_CONVERTER.deserialize(None) is None
+
+
+def test_coerce_plain_dict_to_type():
+    # get_input(expected_type=...) relies on the converter coercing a plain
+    # (already-deserialized) dict into the declared type.
+    coerced = DEFAULT_FUNCTIONS_DATA_CONVERTER.coerce({"x": 5, "y": 6}, Point)
+    assert coerced == Point(5, 6)