Centralize JSON serialization and add type-aware payload encoding/decoding

.github/workflows/validate.yml

@@ -13,19 +13,44 @@ on:
 jobs:
   validate:
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          # Fallback path: on Python 3.9 the SDK's df_dumps / df_loads cannot
+          # be installed (azure-functions 2.x requires >=3.13 and the 1.26.0
+          # line requires >=3.10), so this leg exercises the legacy
+          # serialization fallback in df_serialization.
+          - python-version: "3.9"
+            functions-sdk: ""
+          # SDK path: Python 3.13 with the beta that first ships df_dumps /
+          # df_loads, exercising the SDK-delegated serialization branch.
+          # TODO: change to "azure-functions>=2.2.0" once 2.2.0 GA ships, and
+          # drop the explicit override step below.
+          - python-version: "3.13"
+            functions-sdk: "azure-functions>=2.2.0b5"
     steps:
       - name: Checkout repository
         uses: actions/checkout@v2
 
       - name: Set up Python
         uses: actions/setup-python@v2
         with:
-          python-version: 3.9
+          python-version: ${{ matrix.python-version }}
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
+      - name: Install Functions SDK override
+        if: matrix.functions-sdk != ''
+        run: pip install "${{ matrix.functions-sdk }}"
       - name: Run Linter
+        # Lint only on the canonical Python version. On Python 3.12+, PEP 701
+        # changed f-string tokenization so pycodestyle inspects tokens inside
+        # f-strings, producing false positives (e.g. the ':' in 'http://' or
+        # the indentation of multi-line f-string concatenations). Linting is
+        # environment-agnostic, so running it once on 3.9 is sufficient.
+        if: matrix.python-version == '3.9'
         run: |
           cd azure
           flake8 . --count --show-source --statistics

CHANGELOG.md

@@ -7,6 +7,9 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Client operation correlation logging: `FunctionInvocationId` is now propagated via HTTP headers to the host for client operations, enabling correlation with host logs.
+- Centralized JSON serialization module (`azure.durable_functions.models.utils.df_serialization`): all serialization/deserialization of user payloads (orchestrator inputs/outputs, activity arguments and results, sub-orchestrator payloads, entity inputs/outputs, and client inputs) now flows through `df_dumps` / `df_loads`, replacing scattered `json.dumps(…, default=_serialize_custom_object)` / `json.loads(…, object_hook=_deserialize_custom_object)` calls. This module is a thin shim over the Azure Functions SDK: when the installed `azure-functions` exposes `df_dumps` / `df_loads` (the centralized serializers with type-validation and strict-typing support), they are used directly so our serialization matches the SDK's `ActivityTriggerConverter` at the host boundary; otherwise it falls back to the legacy `_serialize_custom_object` / `_deserialize_custom_object` hooks, which keeps both sides symmetric. The wire format is **unchanged** — builtins serialize to plain JSON and custom objects continue to use the `{"__class__", "__module__", "__data__"}` convention.
+- Type-hint-driven validation via `df_loads(s, expected_type=...)`: when the V2 programming model provides a return-type annotation for an activity or sub-orchestrator, the annotation is threaded through call sites so the SDK's `df_loads` can validate the deserialized payload against that type (when available). On older `azure-functions` releases the argument is accepted but ignored.
+- Return-type discovery for V2 decorated activities/sub-orchestrators (`azure.durable_functions.models.utils.type_discovery`): resolves the concrete return annotation from the user's registered function, used to supply `expected_type` to `df_loads`.
 
 ## 1.0.0b6
 

azure-pipelines.yml

@@ -65,5 +65,33 @@ stages:
       inputs:
         PathtoPublish: dist
         ArtifactName: $(componentArtifactName)
+
+  - job: Test_Functions_Sdk_Path
+    displayName: Test SDK Serialization Path (Py 3.13)
+    # The Build_Durable_Functions job runs on Python 3.9, where the SDK's
+    # df_dumps / df_loads cannot be installed (azure-functions 2.x requires
+    # >=3.13), so it only exercises the legacy serialization fallback.  This
+    # job runs on Python 3.13 with the beta that first ships df_dumps /
+    # df_loads to cover the SDK-delegated branch in df_serialization.
+    # TODO: change the override to 'azure-functions>=2.2.0' once 2.2.0 GA
+    # ships, and drop the explicit install step.
+    pool:
+      name: "1ES-Hosted-AzFunc"
+      demands:
+          - ImageOverride -equals MMSUbuntu20.04TLS
+    steps:
+    - task: UsePythonVersion@0
+      inputs:
+        versionSpec: '3.13'
+    - script: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+        pip install "azure-functions>=2.2.0b5"
+      workingDirectory: $(baseFolder)
+      displayName: 'Install dependencies (SDK serializers)'
+    - script: |
+        pip install pytest pytest-azurepipelines
+        pytest --ignore=samples-v2
+      displayName: 'pytest'
 
 

azure/durable_functions/decorators/durable_app.py

@@ -76,7 +76,7 @@ def decorator(entity_func):
 
         return decorator
 
-    def _configure_orchestrator_callable(self, wrap) -> Callable:
+    def _configure_orchestrator_callable(self, wrap, input_type=None) -> Callable:
         """Obtain decorator to construct an Orchestrator class from a user-defined Function.
 
         In the old programming model, this decorator's logic was unavoidable boilerplate
@@ -86,6 +86,9 @@ def _configure_orchestrator_callable(self, wrap) -> Callable:
         ----------
         wrap: Callable
             The next decorator to be applied.
+        input_type: Optional[type]
+            The expected type for orchestration input, forwarded from
+            the orchestration_trigger decorator.
 
         Returns
         -------
@@ -99,12 +102,16 @@ def decorator(orchestrator_func):
 
             # invoke next decorator, with the Orchestrator as input
             handle.__name__ = orchestrator_func.__name__
+            # Stash the decorator-declared input type so the runtime
+            # can feed it to df_loads via context.get_input().
+            handle._df_input_type = input_type
             return wrap(handle)
 
         return decorator
 
     def orchestration_trigger(self, context_name: str,
-                              orchestration: Optional[str] = None):
+                              orchestration: Optional[str] = None,
+                              input_type: Optional[type] = None):
         """Register an Orchestrator Function.
 
         Parameters
@@ -114,8 +121,13 @@ 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,
+            ``context.get_input()`` will use this type to decode the
+            input payload without consulting ``sys.modules``. A
+            call-site ``expected_type`` argument on ``get_input``
+            takes precedence over this value.
         """
-        @self._configure_orchestrator_callable
         @self._configure_function_builder
         def wrap(fb):
 
@@ -127,7 +139,7 @@ def decorator():
 
             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):

azure/durable_functions/entity.py

@@ -80,6 +80,7 @@ def handle(self, context: DurableEntityContext, batch: List[Dict[str, Any]]) ->
             response.results.append(operation_result)
 
         response.state = context._state
+        response.state_is_raw = context._state_is_raw
         response.entity_exists = context._exists
         return response.to_json_string()
 

azure/durable_functions/models/DurableEntityContext.py

@@ -1,5 +1,5 @@
 from typing import Optional, Any, Dict, Tuple, List, Callable
-from azure.functions._durable_functions import _deserialize_custom_object
+from .utils.df_serialization import df_loads
 import json
 
 
@@ -36,6 +36,7 @@ def __init__(self,
         self._is_newly_constructed: bool = False
 
         self._state: Any = state
+        self._state_is_raw: bool = False
         self._input: Any = None
         self._operation: Optional[str] = None
         self._result: Any = None
@@ -107,12 +108,15 @@ def from_json(cls, json_str: str) -> Tuple['DurableEntityContext', List[Dict[str
         json_dict["key"] = json_dict["self"]["key"]
         json_dict.pop("self")
 
+        # Keep the raw serialized state (a JSON string) so get_state() can
+        # deserialize lazily with an expected_type supplied by the user.
         serialized_state = json_dict["state"]
-        if serialized_state is not None:
-            json_dict["state"] = from_json_util(serialized_state)
 
         batch = json_dict.pop("batch")
-        return cls(**json_dict), batch
+        ctx = cls(**json_dict)
+        if serialized_state is not None:
+            ctx._state_is_raw = True
+        return ctx, batch
 
     def set_state(self, state: Any) -> None:
         """Set the state of the entity.
@@ -126,20 +130,35 @@ def set_state(self, state: Any) -> None:
 
         # should only serialize the state at the end of the batch
         self._state = state
+        # The new state is a live Python value, not the raw JSON string
+        # loaded from the payload. Clear the raw flag so a subsequent
+        # get_state() in the same batch does not try to re-decode it.
+        self._state_is_raw = False
 
-    def get_state(self, initializer: Optional[Callable[[], Any]] = None) -> Any:
+    def get_state(self, initializer: Optional[Callable[[], Any]] = None,
+                  expected_type: Optional[type] = None) -> Any:
         """Get the current state of this entity.
 
         Parameters
         ----------
         initializer: Optional[Callable[[], Any]]
             A 0-argument function to provide an initial state. Defaults to None.
+        expected_type: Optional[type]
+            The type to decode the state as. When set, the codec uses
+            this type directly without consulting ``sys.modules``. Note that
+            the persisted state is decoded lazily on the **first** get_state
+            call within a batch; an ``expected_type`` supplied on a later
+            call (after the state has already been decoded or replaced via
+            set_state) has no effect.
 
         Returns
         -------
         Any
             The current state of the entity
         """
+        if self._state is not None and self._state_is_raw:
+            self._state = from_json_util(self._state, expected_type=expected_type)
+            self._state_is_raw = False
         state = self._state
         if state is not None:
             return state
@@ -149,9 +168,15 @@ def get_state(self, initializer: Optional[Callable[[], Any]] = None) -> Any:
             state = initializer()
         return state
 
-    def get_input(self) -> Any:
+    def get_input(self, expected_type: Optional[type] = None) -> Any:
         """Get the input for this operation.
 
+        Parameters
+        ----------
+        expected_type: Optional[type]
+            The type to decode the input as. When set, the codec uses
+            this type directly without consulting ``sys.modules``.
+
         Returns
         -------
         Any
@@ -160,7 +185,7 @@ def get_input(self) -> Any:
         input_ = None
         req_input = self._input
         req_input = json.loads(req_input)
-        input_ = None if req_input is None else from_json_util(req_input)
+        input_ = None if req_input is None else df_loads(req_input, expected_type=expected_type)
         return input_
 
     def set_result(self, result: Any) -> None:
@@ -178,9 +203,10 @@ def destruct_on_exit(self) -> None:
         """Delete this entity after the operation completes."""
         self._exists = False
         self._state = None
+        self._state_is_raw = False
 
 
-def from_json_util(json_str: str) -> Any:
+def from_json_util(json_str: str, expected_type: Optional[type] = None) -> Any:
     """Load an arbitrary datatype from its JSON representation.
 
     The Out-of-proc SDK has a special JSON encoding strategy
@@ -192,10 +218,12 @@ def from_json_util(json_str: str) -> Any:
     ----------
     json_str: str
         A JSON-formatted string, from durable-extension
+    expected_type: Optional[type]
+        The type to decode the value as.
 
     Returns
     -------
     Any:
         The original datatype that was serialized
     """
-    return json.loads(json_str, object_hook=_deserialize_custom_object)
+    return df_loads(json_str, expected_type=expected_type)

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -16,7 +16,7 @@
 from ..models.DurableOrchestrationBindings import DurableOrchestrationBindings
 from .utils.http_utils import get_async_request, post_async_request, delete_async_request
 from .utils.entity_utils import EntityId
-from azure.functions._durable_functions import _serialize_custom_object
+from .utils.df_serialization import df_dumps
 
 
 class DurableOrchestrationClient:
@@ -633,7 +633,7 @@ def _get_json_input(client_input: object) -> Optional[str]:
             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 df_dumps(client_input)
         return None
 
     @staticmethod