Serialization improvements (WIP)

Author: andystaples

CHANGELOG.md

@@ -27,7 +27,12 @@ ADDED
   custom status, entity state) routes through it. The default
   `JsonDataConverter` preserves existing behavior, so a custom converter (for
   example one backed by pydantic) is opt-in. Custom objects can opt in via a
-  `to_json()` hook and a `from_json(value)` classmethod.
+  `to_json()` hook and a `from_json(value)` classmethod. Objects that contain
+  other hook-using objects round-trip automatically: nested `to_json()` hooks
+  fire at any depth during serialization, and a `from_json` hook may declare an
+  optional second parameter (`from_json(cls, value, converter)`) to reconstruct
+  nested typed values via `converter.coerce(child, ChildType)` instead of by
+  hand.
 - `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, the result/event payload is

durabletask/internal/json_codec.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import dataclasses
+import inspect
 import json
 import types
 import typing
@@ -47,7 +48,8 @@ def to_json(obj: Any) -> str:
         ) from e
 
 
-def from_json(json_str: str | bytes | bytearray, expected_type: type | None = None) -> Any:
+def from_json(json_str: str | bytes | bytearray, expected_type: type | None = None,
+              converter: Any | None = None) -> Any:
     """Deserialize a JSON string, optionally coercing the result to a type.
 
     When ``expected_type`` is ``None`` (the default) the raw parsed JSON is
@@ -62,11 +64,15 @@ def from_json(json_str: str | bytes | bytearray, expected_type: type | None = No
     classmethod are reconstructed via that hook, and ``Optional``/``Union`` and
     ``list`` type hints are honored recursively. The destination type is always
     supplied by the caller; it is never read from the payload.
+
+    ``converter`` is the active :class:`~durabletask.serialization.DataConverter`.
+    It is forwarded to ``from_json`` hooks that opt in (see :func:`coerce_to_type`)
+    so they can reconstruct nested typed values via ``converter.coerce(...)``.
     """
     if expected_type is None:
         return json.loads(json_str, object_hook=_legacy_object_hook)
     raw = json.loads(json_str, object_hook=_strip_legacy_marker)
-    return coerce_to_type(raw, expected_type)
+    return coerce_to_type(raw, expected_type, converter)
 
 
 def _encode_custom_object(o: Any) -> Any:
@@ -92,7 +98,13 @@ def _encode_custom_object(o: Any) -> Any:
     if callable(to_json_hook):
         return to_json_hook(o)
     if dataclasses.is_dataclass(o) and not isinstance(o, type):
-        return dataclasses.asdict(o)
+        # Shallow-convert to a dict whose *values are the original field objects*
+        # (unlike ``dataclasses.asdict``, which deep-recurses and would convert a
+        # nested dataclass via ``asdict`` -- bypassing that child's ``to_json``
+        # hook). ``json.dumps`` then recurses into each value and re-enters this
+        # hook for any nested custom object, so nested ``to_json`` hooks fire at
+        # every depth (including inside lists/dicts).
+        return {f.name: getattr(o, f.name) for f in dataclasses.fields(o)}
     if isinstance(o, SimpleNamespace):
         return vars(o)
     # This will raise a TypeError describing the unsupported type.
@@ -112,20 +124,31 @@ def _strip_legacy_marker(d: dict[str, Any]) -> dict[str, Any]:
     return d
 
 
-def coerce_to_type(value: Any, expected_type: Any) -> Any:
+def coerce_to_type(value: Any, expected_type: Any, converter: Any | None = None) -> Any:
     """Coerce an already-parsed JSON value to ``expected_type``.
 
     Handles ``None``/``Optional``/``Union`` and ``list`` type hints recursively,
     types exposing a ``from_json()`` classmethod, and dataclasses (including
     nested dataclass fields). The destination type is always caller-supplied and
     never derived from the payload, keeping deserialization secure.
+
+    ``converter`` is the active :class:`~durabletask.serialization.DataConverter`.
+    A ``from_json`` hook may opt in to receiving it (by accepting a second
+    positional parameter) and delegate nested reconstruction back to the
+    converter, e.g. ``converter.coerce(child, ChildType)``. This keeps hooks free
+    of manual nested deserialization and routes children through the same policy.
     """
     if expected_type is None or value is None:
         return value
 
+    # ``Any`` imposes no constraint -- and ``isinstance(x, Any)`` raises -- so
+    # short-circuit before any type inspection below.
+    if expected_type is typing.Any:
+        return value
+
     origin = typing.get_origin(expected_type)
     if origin is not None:
-        return _coerce_generic(value, expected_type, origin)
+        return _coerce_generic(value, expected_type, origin, converter)
 
     if not isinstance(expected_type, type):
         # Not a concrete, instantiable type (e.g. a typing special form we don't
@@ -137,10 +160,10 @@ def coerce_to_type(value: Any, expected_type: Any) -> Any:
 
     from_json_hook = getattr(expected_type, "from_json", None)
     if callable(from_json_hook):
-        return from_json_hook(value)
+        return _invoke_from_json(from_json_hook, value, converter)
 
     if dataclasses.is_dataclass(expected_type) and isinstance(value, dict):
-        return _build_dataclass(expected_type, cast(dict[str, Any], value))
+        return _build_dataclass(expected_type, cast(dict[str, Any], value), converter)
 
     type_ctor = cast(Any, expected_type)
     try:
@@ -153,29 +176,69 @@ def coerce_to_type(value: Any, expected_type: Any) -> Any:
         ) from e
 
 
-def _coerce_generic(value: Any, expected_type: Any, origin: Any) -> Any:
+def _invoke_from_json(from_json_hook: Any, value: Any, converter: Any | None) -> Any:
+    """Invoke a ``from_json`` hook, passing the converter if the hook accepts it.
+
+    Hooks may be declared as ``from_json(cls, value)`` (the original contract) or
+    ``from_json(cls, value, converter)`` to opt into managed nested
+    reconstruction. Arity is detected from the bound hook's signature, so the
+    extra parameter is fully backwards compatible. When a converter-aware hook is
+    found but no converter was threaded (e.g. a direct ``json_codec`` call), the
+    shared default converter is resolved lazily so the hook always receives one.
+    """
+    wants_converter = False
+    try:
+        params = [
+            p for p in inspect.signature(from_json_hook).parameters.values()
+            if p.kind in (inspect.Parameter.POSITIONAL_ONLY,
+                          inspect.Parameter.POSITIONAL_OR_KEYWORD)
+        ]
+        wants_converter = len(params) >= 2
+    except (TypeError, ValueError):
+        wants_converter = False
+
+    if wants_converter:
+        if converter is None:
+            converter = _default_converter()
+        return from_json_hook(value, converter)
+    return from_json_hook(value)
+
+
+def _default_converter() -> Any:
+    # Lazy import to avoid a load-time cycle: ``serialization`` imports this
+    # module at import time, but by the time a hook actually runs both modules
+    # are fully initialized.
+    from durabletask.serialization import DEFAULT_DATA_CONVERTER
+    return DEFAULT_DATA_CONVERTER
+
+
+def _coerce_generic(value: Any, expected_type: Any, origin: Any, converter: Any | None = None) -> Any:
     args = typing.get_args(expected_type)
     if origin is typing.Union or origin is types.UnionType:
         # If the value already matches a member type, keep it as-is.
         non_none = [a for a in args if a is not type(None)]
         for arg in non_none:
+            # An ``Any`` member imposes no constraint (and ``isinstance(x, Any)``
+            # raises), so the value is acceptable as-is.
+            if arg is typing.Any:
+                return value
             if isinstance(arg, type) and isinstance(value, arg):
                 return value
         # ``Optional[T]`` (exactly one non-None member): coerce to that member.
         # For a genuine multi-member ``Union`` where the value matched none of
         # the members, leave it untouched rather than guessing the first arg --
         # forcing a coercion there can silently mis-construct the wrong type.
         if len(non_none) == 1:
-            return coerce_to_type(value, non_none[0])
+            return coerce_to_type(value, non_none[0], converter)
         return value
     if origin in (list, Sequence) and isinstance(value, list):
         elem_type = args[0] if args else None
-        return [coerce_to_type(item, elem_type) for item in cast(list[Any], value)]
+        return [coerce_to_type(item, elem_type, converter) for item in cast(list[Any], value)]
     # Other generics (dict, tuple, ...) are returned as parsed JSON.
     return value
 
 
-def _build_dataclass(cls: Any, data: dict[str, Any]) -> Any:
+def _build_dataclass(cls: Any, data: dict[str, Any], converter: Any | None = None) -> Any:
     """Construct a dataclass from its dict payload, recursing into typed fields."""
     try:
         hints = typing.get_type_hints(cls)
@@ -186,5 +249,5 @@ def _build_dataclass(cls: Any, data: dict[str, Any]) -> Any:
         if field.name not in data:
             continue
         field_type = hints.get(field.name)
-        kwargs[field.name] = coerce_to_type(data[field.name], field_type)
+        kwargs[field.name] = coerce_to_type(data[field.name], field_type, converter)
     return cls(**kwargs)

durabletask/scheduled/client.py

@@ -2,13 +2,10 @@
 # Licensed under the MIT License.
 
 import logging
-from dataclasses import asdict
-from typing import Any
 
 from durabletask.client import (EntityQuery, OrchestrationStatus,
                                 TaskHubGrpcClient)
 from durabletask.entities import EntityInstanceId
-from durabletask.internal import shared
 from durabletask.scheduled import transitions
 from durabletask.scheduled.exceptions import ScheduleNotFoundError
 from durabletask.scheduled.models import (ScheduleCreationOptions,
@@ -22,20 +19,6 @@
 logger = logging.getLogger("durabletask.scheduled")
 
 
-def _parse_state(serialized_state: Any) -> ScheduleState | None:
-    if serialized_state is None:
-        return None
-    data = serialized_state
-    if isinstance(data, str):
-        if not data.strip():
-            # A deleted (or never-initialized) entity reports empty state.
-            return None
-        data = shared.from_json(data)
-    if isinstance(data, dict):
-        return ScheduleState.from_dict(data)
-    return None
-
-
 class ScheduleClient:
     """Client for managing a single schedule instance."""
 
@@ -53,14 +36,14 @@ def schedule_id(self) -> str:
         """Gets the ID of this schedule."""
         return self._schedule_id
 
-    def _run_operation(self, operation_name: str, input: Any | None = None) -> None:
+    def _run_operation(self, operation_name: str, input: object | None = None) -> None:
         request = ScheduleOperationRequest(
             entity_id=str(self._entity_id),
             operation_name=operation_name,
             input=input,
         )
         instance_id = self._client.schedule_new_orchestration(
-            execute_schedule_operation_orchestrator, input=asdict(request))
+            execute_schedule_operation_orchestrator, input=request)
         state = self._client.wait_for_orchestration_completion(
             instance_id, timeout=self._operation_timeout)
         if state is None or state.runtime_status != OrchestrationStatus.COMPLETED:
@@ -71,11 +54,11 @@ def _run_operation(self, operation_name: str, input: Any | None = None) -> None:
 
     def create(self, options: ScheduleCreationOptions) -> None:
         """Create or update this schedule with the given configuration."""
-        self._run_operation(transitions.CREATE_SCHEDULE, options.to_json())
+        self._run_operation(transitions.CREATE_SCHEDULE, options)
 
     def update(self, options: ScheduleUpdateOptions) -> None:
         """Update this schedule's configuration."""
-        self._run_operation(transitions.UPDATE_SCHEDULE, options.to_json())
+        self._run_operation(transitions.UPDATE_SCHEDULE, options)
 
     def pause(self) -> None:
         """Pause this schedule."""
@@ -94,7 +77,7 @@ def describe(self) -> ScheduleDescription:
         metadata = self._client.get_entity(self._entity_id, include_state=True)
         if metadata is None:
             raise ScheduleNotFoundError(self._schedule_id)
-        state = _parse_state(metadata.get_state())
+        state = metadata.get_typed_state(ScheduleState)
         if state is None:
             raise ScheduleNotFoundError(self._schedule_id)
         return state.to_description()
@@ -136,7 +119,7 @@ def list_schedules(self, filter: ScheduleQuery | None = None) -> list[ScheduleDe
         )
         results: list[ScheduleDescription] = []
         for metadata in self._client.get_all_entities(query):
-            state = _parse_state(metadata.get_state())
+            state = metadata.get_typed_state(ScheduleState)
             if state is None or state.schedule_configuration is None:
                 continue
             if not self._matches_filter(state, filter):

durabletask/scheduled/models.py

@@ -220,7 +220,7 @@ def _validate(self):
         if self.start_at is not None and self.end_at is not None and self.start_at > self.end_at:
             raise ValueError("start_at cannot be later than end_at.")
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_json(self) -> dict[str, Any]:
         return {
             "schedule_id": self.schedule_id,
             "orchestration_name": self.orchestration_name,
@@ -232,9 +232,9 @@ def to_dict(self) -> dict[str, Any]:
             "start_immediately_if_late": self.start_immediately_if_late,
         }
 
-    @staticmethod
-    def from_dict(data: dict[str, Any]) -> "ScheduleConfiguration":
-        config = ScheduleConfiguration(
+    @classmethod
+    def from_json(cls, data: dict[str, Any]) -> "ScheduleConfiguration":
+        config = cls(
             data["schedule_id"],
             data["orchestration_name"],
             timedelta(seconds=data["interval_seconds"]),
@@ -262,29 +262,35 @@ def __init__(self):
     def refresh_execution_token(self):
         self.execution_token = _new_token()
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_json(self) -> dict[str, Any]:
+        # ``schedule_configuration`` is returned as the object itself; the
+        # serializer recurses into it and fires its own ``to_json`` hook. Only
+        # this type's non-JSON-native leaves (datetimes) are converted here.
         return {
             "status": self.status.value,
             "execution_token": self.execution_token,
             "last_run_at": _to_iso(self.last_run_at),
             "next_run_at": _to_iso(self.next_run_at),
             "schedule_created_at": _to_iso(self.schedule_created_at),
             "schedule_last_modified_at": _to_iso(self.schedule_last_modified_at),
-            "schedule_configuration":
-                self.schedule_configuration.to_dict() if self.schedule_configuration else None,
+            "schedule_configuration": self.schedule_configuration,
         }
 
-    @staticmethod
-    def from_dict(data: dict[str, Any]) -> "ScheduleState":
-        state = ScheduleState()
+    @classmethod
+    def from_json(cls, data: dict[str, Any], converter: Any) -> "ScheduleState":
+        # The nested configuration is reconstructed through the converter, which
+        # routes it to ``ScheduleConfiguration``'s own ``from_json`` hook (and
+        # honors a custom converter). Only this type's datetime leaves are
+        # rebuilt by hand.
+        state = cls()
         state.status = ScheduleStatus(data["status"])
         state.execution_token = data["execution_token"]
         state.last_run_at = _from_iso(data.get("last_run_at"))
         state.next_run_at = _from_iso(data.get("next_run_at"))
         state.schedule_created_at = _from_iso(data.get("schedule_created_at"))
         state.schedule_last_modified_at = _from_iso(data.get("schedule_last_modified_at"))
-        config_data = data.get("schedule_configuration")
-        state.schedule_configuration = ScheduleConfiguration.from_dict(config_data) if config_data else None
+        state.schedule_configuration = converter.coerce(
+            data.get("schedule_configuration"), ScheduleConfiguration)
         return state
 
     def to_description(self) -> ScheduleDescription:

durabletask/scheduled/orchestrator.py

@@ -3,7 +3,6 @@
 
 from collections.abc import Generator
 from dataclasses import dataclass
-from types import SimpleNamespace
 from typing import Any
 
 from durabletask import task
@@ -12,25 +11,27 @@
 
 @dataclass
 class ScheduleOperationRequest:
-    """Request describing an operation to execute against a schedule entity."""
+    """Request describing an operation to execute against a schedule entity.
+
+    A plain dataclass: the serializer round-trips it (and its ``input`` payload)
+    automatically. ``input`` stays an ``Any`` here -- it is reconstructed into the
+    concrete options type at the entity-method boundary from that method's
+    parameter annotation.
+    """
 
     entity_id: str
     operation_name: str
     input: Any | None = None
 
 
 def execute_schedule_operation_orchestrator(
-        ctx: task.OrchestrationContext, request: Any) -> Generator[task.Task[Any], Any, Any]:
+        ctx: task.OrchestrationContext,
+        request: ScheduleOperationRequest) -> Generator[task.Task[Any], Any, Any]:
     """Orchestrator that executes a single operation on a schedule entity.
 
     Client-side write operations route through this orchestrator so callers can await
     completion (and surface failures) of the underlying entity operation.
     """
-    if isinstance(request, SimpleNamespace):
-        request = vars(request)
-    if isinstance(request, dict):
-        request = ScheduleOperationRequest(**request)
-
     entity_id = EntityInstanceId.parse(request.entity_id)
     result = yield ctx.call_entity(entity_id, request.operation_name, request.input)
     return result