PR Feedback (DataConverter pattern, small fixes)

Author: andystaples

CHANGELOG.md

@@ -52,6 +52,17 @@ ADDED
   uses a deterministic instance ID (`export-job-{job_id}`, exposed via
   `orchestrator_instance_id_for(...)`) so callers can correlate a job ID
   with its orchestrator for logging, monitoring, and restart.
+- Added a pluggable `DataConverter` abstraction (`durabletask.serialization`)
+  consumed by both the worker and the client. `TaskHubGrpcWorker`,
+  `TaskHubGrpcClient`, and `AsyncTaskHubGrpcClient` accept a `data_converter`
+  argument; every payload serialization boundary (inputs, outputs, events,
+  custom status, entity state) routes through it. The default
+  `JsonDataConverter` preserves existing behavior, so supplying a custom
+  converter (for example, one backed by pydantic) is purely opt-in. Custom
+  objects opt in via a `to_json()` hook -- invoked as `type(obj).to_json(obj)`
+  so both instance methods and `@staticmethod` hooks work -- and a
+  `from_json(value)` classmethod, matching the `azure-functions-durable`
+  convention.
 - Type-aware deserialization of payloads. `OrchestrationContext.call_activity`,
   `call_sub_orchestrator`, and `call_entity` accept an optional `return_type`,
   and `wait_for_external_event` accepts an optional `data_type`. When provided,
@@ -65,9 +76,11 @@ ADDED
 - Inbound payloads are reconstructed from function type annotations. When an
   orchestrator, activity, or entity operation annotates its input parameter with
   a dataclass or a `from_json()`-capable type, the incoming payload is
-  automatically coerced to that type. Discovery is best-effort and conservative:
-  builtins and unannotated/unknown types are passed through unchanged, and a
-  payload that cannot be coerced falls back to the raw value.
+  automatically coerced to that type. Coercion is best-effort: builtins and
+  unannotated/unknown types are passed through unchanged, and a payload that
+  cannot be coerced to the requested type falls back to the raw deserialized
+  value (logged at debug level) rather than raising. This best-effort policy is
+  owned by the default `JsonDataConverter`; a stricter converter can change it.
 - `call_activity` results are reconstructed from the activity's return
   annotation. When an activity function reference is passed (not a string name)
   and its return type is annotated with a dataclass or `from_json()`-capable

durabletask/client.py

@@ -7,7 +7,7 @@
 import time
 import uuid
 from collections.abc import AsyncIterable, Iterable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 from typing import Any, Generic, Protocol, TypeVar, cast, overload
@@ -50,6 +50,7 @@
 )
 from durabletask.payload import helpers as payload_helpers
 from durabletask.payload.store import PayloadStore
+from durabletask.serialization import DEFAULT_DATA_CONVERTER, DataConverter, JsonDataConverter
 
 TInput = TypeVar('TInput')
 TOutput = TypeVar('TOutput')
@@ -82,6 +83,12 @@ class OrchestrationState:
     serialized_output: str | None
     serialized_custom_status: str | None
     failure_details: task.FailureDetails | None
+    # Converter used by the typed accessors below. Defaults to the SDK's JSON
+    # converter; the client populates it with its own converter so custom
+    # serialization applies on the read side too. Excluded from equality/repr so
+    # two states with equal payloads remain equal regardless of converter.
+    _data_converter: DataConverter = field(
+        default=DEFAULT_DATA_CONVERTER, compare=False, repr=False)
 
     @overload
     def get_input(self, expected_type: type[T]) -> T | None:
@@ -111,7 +118,7 @@ def get_input(self, expected_type: type | None = None) -> Any:
         """
         if self.serialized_input is None:
             return None
-        return shared.from_json(self.serialized_input, expected_type)
+        return self._data_converter.deserialize(self.serialized_input, expected_type)
 
     @overload
     def get_output(self, expected_type: type[T]) -> T | None:
@@ -141,7 +148,7 @@ def get_output(self, expected_type: type | None = None) -> Any:
         """
         if self.serialized_output is None:
             return None
-        return shared.from_json(self.serialized_output, expected_type)
+        return self._data_converter.deserialize(self.serialized_output, expected_type)
 
     @overload
     def get_custom_status(self, expected_type: type[T]) -> T | None:
@@ -171,7 +178,7 @@ def get_custom_status(self, expected_type: type | None = None) -> Any:
         """
         if self.serialized_custom_status is None:
             return None
-        return shared.from_json(self.serialized_custom_status, expected_type)
+        return self._data_converter.deserialize(self.serialized_custom_status, expected_type)
 
     def raise_if_failed(self):
         if self.failure_details is not None:
@@ -229,18 +236,22 @@ def failure_details(self):
         return self._failure_details
 
 
-def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> OrchestrationState | None:
+def new_orchestration_state(
+        instance_id: str, res: pb.GetInstanceResponse,
+        data_converter: DataConverter | None = None) -> OrchestrationState | None:
     if not res.exists:
         return None
 
     state = res.orchestrationState
 
-    new_state = parse_orchestration_state(state)
+    new_state = parse_orchestration_state(state, data_converter)
     new_state.instance_id = instance_id  # Override instance_id with the one from the request, to match old behavior
     return new_state
 
 
-def parse_orchestration_state(state: pb.OrchestrationState) -> OrchestrationState:
+def parse_orchestration_state(
+        state: pb.OrchestrationState,
+        data_converter: DataConverter | None = None) -> OrchestrationState:
     failure_details = None
     if state.failureDetails.errorMessage != '' or state.failureDetails.errorType != '':
         failure_details = task.FailureDetails(
@@ -257,7 +268,8 @@ def parse_orchestration_state(state: pb.OrchestrationState) -> OrchestrationStat
         state.input.value if not helpers.is_empty(state.input) else None,
         state.output.value if not helpers.is_empty(state.output) else None,
         state.customStatus.value if not helpers.is_empty(state.customStatus) else None,
-        failure_details)
+        failure_details,
+        data_converter if data_converter is not None else DEFAULT_DATA_CONVERTER)
 
 
 # Grace period before a retired SDK-owned channel is force-closed. Long enough
@@ -400,9 +412,11 @@ def __init__(self, *,
                  channel_options: GrpcChannelOptions | None = None,
                  resiliency_options: GrpcClientResiliencyOptions | None = None,
                  default_version: str | None = None,
-                 payload_store: PayloadStore | None = None):
+                 payload_store: PayloadStore | None = None,
+                 data_converter: DataConverter | None = None):
 
         self._owns_channel = channel is None
+        self._data_converter = data_converter if data_converter is not None else JsonDataConverter()
         self._host_address = (
             host_address if host_address else shared.get_default_host_address()
         )
@@ -605,7 +619,8 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu
             req = build_schedule_new_orchestration_req(
                 orchestrator, input=input, instance_id=instance_id, start_at=start_at,
                 reuse_id_policy=reuse_id_policy, tags=tags,
-                version=version if version else self.default_version)
+                version=version if version else self.default_version,
+                data_converter=self._data_converter)
 
             # Inject the active PRODUCER span context into the request so the sidecar
             # stores it in the executionStarted event and the worker can parent all
@@ -629,7 +644,7 @@ def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = Tr
         # De-externalize any large-payload tokens in the response
         if self._payload_store is not None and res.exists:
             payload_helpers.deexternalize_payloads(res, self._payload_store)
-        return new_orchestration_state(req.instanceId, res)
+        return new_orchestration_state(req.instanceId, res, self._data_converter)
 
     def get_orchestration_history(self,
                                   instance_id: str, *,
@@ -685,7 +700,7 @@ def get_all_orchestration_states(self,
             resp: pb.QueryInstancesResponse = self._stub.QueryInstances(req)
             if self._payload_store is not None:
                 payload_helpers.deexternalize_payloads(resp, self._payload_store)
-            states += [parse_orchestration_state(res) for res in resp.orchestrationState]
+            states += [parse_orchestration_state(res, self._data_converter) for res in resp.orchestrationState]
             if check_continuation_token(resp.continuationToken, _continuation_token, self._logger):
                 _continuation_token = resp.continuationToken
             else:
@@ -705,7 +720,7 @@ def wait_for_orchestration_start(self, instance_id: str, *,
             )
             if self._payload_store is not None and res.exists:
                 payload_helpers.deexternalize_payloads(res, self._payload_store)
-            return new_orchestration_state(req.instanceId, res)
+            return new_orchestration_state(req.instanceId, res, self._data_converter)
         except grpc.RpcError as rpc_error:
             if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:  # type: ignore
                 # Replace gRPC error with the built-in TimeoutError
@@ -725,7 +740,7 @@ def wait_for_orchestration_completion(self, instance_id: str, *,
             )
             if self._payload_store is not None and res.exists:
                 payload_helpers.deexternalize_payloads(res, self._payload_store)
-            state = new_orchestration_state(req.instanceId, res)
+            state = new_orchestration_state(req.instanceId, res, self._data_converter)
             log_completion_state(self._logger, instance_id, state)
             return state
         except grpc.RpcError as rpc_error:
@@ -737,7 +752,7 @@ def wait_for_orchestration_completion(self, instance_id: str, *,
     def raise_orchestration_event(self, instance_id: str, event_name: str, *,
                                   data: Any | None = None) -> None:
         with tracing.start_raise_event_span(event_name, instance_id):
-            req = build_raise_event_req(instance_id, event_name, data)
+            req = build_raise_event_req(instance_id, event_name, data, self._data_converter)
             self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.")
             if self._payload_store is not None:
                 payload_helpers.externalize_payloads(
@@ -748,7 +763,7 @@ def raise_orchestration_event(self, instance_id: str, event_name: str, *,
     def terminate_orchestration(self, instance_id: str, *,
                                 output: Any | None = None,
                                 recursive: bool = True) -> None:
-        req = build_terminate_req(instance_id, output, recursive)
+        req = build_terminate_req(instance_id, output, recursive, self._data_converter)
 
         self._logger.info(f"Terminating instance '{instance_id}'.")
         if self._payload_store is not None:
@@ -811,7 +826,7 @@ def signal_entity(self,
                       entity_instance_id: EntityInstanceId,
                       operation_name: str,
                       input: Any | None = None) -> None:
-        req = build_signal_entity_req(entity_instance_id, operation_name, input)
+        req = build_signal_entity_req(entity_instance_id, operation_name, input, self._data_converter)
         self._logger.info(f"Signaling entity '{entity_instance_id}' operation '{operation_name}'.")
         if self._payload_store is not None:
             payload_helpers.externalize_payloads(
@@ -896,9 +911,11 @@ def __init__(self, *,
                  channel_options: GrpcChannelOptions | None = None,
                  resiliency_options: GrpcClientResiliencyOptions | None = None,
                  default_version: str | None = None,
-                 payload_store: PayloadStore | None = None):
+                 payload_store: PayloadStore | None = None,
+                 data_converter: DataConverter | None = None):
 
         self._owns_channel = channel is None
+        self._data_converter = data_converter if data_converter is not None else JsonDataConverter()
         self._host_address = (
             host_address if host_address else shared.get_default_host_address()
         )
@@ -1089,7 +1106,8 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu
             req = build_schedule_new_orchestration_req(
                 orchestrator, input=input, instance_id=instance_id, start_at=start_at,
                 reuse_id_policy=reuse_id_policy, tags=tags,
-                version=version if version else self.default_version)
+                version=version if version else self.default_version,
+                data_converter=self._data_converter)
 
             parent_trace_ctx = tracing.get_current_trace_context()
             if parent_trace_ctx is not None:
@@ -1110,7 +1128,7 @@ async def get_orchestration_state(self, instance_id: str, *,
         res: pb.GetInstanceResponse = await self._stub.GetInstance(req)
         if self._payload_store is not None and res.exists:
             await payload_helpers.deexternalize_payloads_async(res, self._payload_store)
-        return new_orchestration_state(req.instanceId, res)
+        return new_orchestration_state(req.instanceId, res, self._data_converter)
 
     async def get_orchestration_history(self,
                                         instance_id: str, *,
@@ -1166,7 +1184,7 @@ async def get_all_orchestration_states(self,
             resp: pb.QueryInstancesResponse = await self._stub.QueryInstances(req)
             if self._payload_store is not None:
                 await payload_helpers.deexternalize_payloads_async(resp, self._payload_store)
-            states += [parse_orchestration_state(res) for res in resp.orchestrationState]
+            states += [parse_orchestration_state(res, self._data_converter) for res in resp.orchestrationState]
             if check_continuation_token(resp.continuationToken, _continuation_token, self._logger):
                 _continuation_token = resp.continuationToken
             else:
@@ -1186,7 +1204,7 @@ async def wait_for_orchestration_start(self, instance_id: str, *,
             )
             if self._payload_store is not None and res.exists:
                 await payload_helpers.deexternalize_payloads_async(res, self._payload_store)
-            return new_orchestration_state(req.instanceId, res)
+            return new_orchestration_state(req.instanceId, res, self._data_converter)
         except grpc.aio.AioRpcError as rpc_error:
             if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
                 raise TimeoutError("Timed-out waiting for the orchestration to start")
@@ -1205,7 +1223,7 @@ async def wait_for_orchestration_completion(self, instance_id: str, *,
             )
             if self._payload_store is not None and res.exists:
                 await payload_helpers.deexternalize_payloads_async(res, self._payload_store)
-            state = new_orchestration_state(req.instanceId, res)
+            state = new_orchestration_state(req.instanceId, res, self._data_converter)
             log_completion_state(self._logger, instance_id, state)
             return state
         except grpc.aio.AioRpcError as rpc_error:
@@ -1217,7 +1235,7 @@ async def wait_for_orchestration_completion(self, instance_id: str, *,
     async def raise_orchestration_event(self, instance_id: str, event_name: str, *,
                                         data: Any | None = None) -> None:
         with tracing.start_raise_event_span(event_name, instance_id):
-            req = build_raise_event_req(instance_id, event_name, data)
+            req = build_raise_event_req(instance_id, event_name, data, self._data_converter)
             self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.")
             if self._payload_store is not None:
                 await payload_helpers.externalize_payloads_async(
@@ -1228,7 +1246,7 @@ async def raise_orchestration_event(self, instance_id: str, event_name: str, *,
     async def terminate_orchestration(self, instance_id: str, *,
                                       output: Any | None = None,
                                       recursive: bool = True) -> None:
-        req = build_terminate_req(instance_id, output, recursive)
+        req = build_terminate_req(instance_id, output, recursive, self._data_converter)
 
         self._logger.info(f"Terminating instance '{instance_id}'.")
         if self._payload_store is not None:
@@ -1291,7 +1309,7 @@ async def signal_entity(self,
                             entity_instance_id: EntityInstanceId,
                             operation_name: str,
                             input: Any | None = None) -> None:
-        req = build_signal_entity_req(entity_instance_id, operation_name, input)
+        req = build_signal_entity_req(entity_instance_id, operation_name, input, self._data_converter)
         self._logger.info(f"Signaling entity '{entity_instance_id}' operation '{operation_name}'.")
         if self._payload_store is not None:
             await payload_helpers.externalize_payloads_async(

durabletask/entities/entity_context.py

@@ -1,20 +1,28 @@
 
-from typing import Any, TypeVar, overload
+from typing import TYPE_CHECKING, Any, TypeVar, overload
 import uuid
 from durabletask.entities.entity_instance_id import EntityInstanceId
-from durabletask.internal import helpers, shared
+from durabletask.internal import helpers
 from durabletask.internal.entity_state_shim import StateShim
 import durabletask.internal.orchestrator_service_pb2 as pb
 
+if TYPE_CHECKING:
+    from durabletask.serialization import DataConverter
+
 TState = TypeVar("TState")
 
 
 class EntityContext:
-    def __init__(self, orchestration_id: str, operation: str, state: StateShim, entity_id: EntityInstanceId):
+    def __init__(self, orchestration_id: str, operation: str, state: StateShim,
+                 entity_id: EntityInstanceId, data_converter: "DataConverter | None" = None):
         self._orchestration_id = orchestration_id
         self._operation = operation
         self._state = state
         self._entity_id = entity_id
+        if data_converter is None:
+            from durabletask.serialization import JsonDataConverter
+            data_converter = JsonDataConverter()
+        self._data_converter = data_converter
 
     @property
     def orchestration_id(self) -> str:
@@ -93,7 +101,7 @@ def signal_entity(self, entity_instance_id: EntityInstanceId, operation: str, in
         input : Any, optional
             The input to provide to the entity for the operation.
         """
-        encoded_input: str | None = shared.to_json(input) if input is not None else None
+        encoded_input: str | None = self._data_converter.serialize(input)
         self._state.add_operation_action(
             pb.OperationAction(
                 sendSignal=pb.SendSignalAction(
@@ -124,7 +132,7 @@ def schedule_new_orchestration(self, orchestration_name: str, input: Any | None
         str
             The instance ID of the scheduled orchestration.
         """
-        encoded_input: str | None = shared.to_json(input) if input is not None else None
+        encoded_input: str | None = self._data_converter.serialize(input)
         if not instance_id:
             instance_id = uuid.uuid4().hex
         self._state.add_operation_action(