Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 100 additions & 14 deletions durabletask/serialization.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@
import types
import typing
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from collections.abc import Set as AbstractSet
from types import SimpleNamespace
from typing import Any, cast

Expand All @@ -68,7 +69,6 @@ class DataConverter(ABC):
@abstractmethod
def serialize(self, value: Any) -> str | None:
"""Serialize ``value`` to a string, or ``None`` when ``value`` is ``None``."""
pass

@abstractmethod
def deserialize(self, data: str | None, target_type: type | None = None) -> Any:
Expand All @@ -85,7 +85,6 @@ def deserialize(self, data: str | None, target_type: type | None = None) -> Any:
:class:`JsonDataConverter` is best-effort and falls back; a validating
converter may instead raise.
"""
pass

@abstractmethod
def coerce(self, value: Any, target_type: type | None = None) -> Any:
Expand All @@ -99,7 +98,6 @@ def coerce(self, value: Any, target_type: type | None = None) -> Any:
(strict vs. best-effort) that an implementation applies in
:meth:`deserialize` should apply here.
"""
pass

def can_reconstruct(self, target_type: Any) -> bool:
"""Return True if this converter can rebuild ``target_type`` from a payload.
Expand Down Expand Up @@ -136,7 +134,9 @@ class JsonDataConverter(DataConverter):
``SimpleNamespace`` handling, and nested values are encoded recursively, so
a dataclass (or any object) with a ``to_json`` hook -- including one nested
inside another dataclass, ``list``, or ``dict`` -- round-trips through its
own hooks.
own hooks. Sets and frozensets serialize to JSON arrays ordered
deterministically by their canonical JSON representation (using lexical
string order, so numeric sets like ``{2, 10}`` serialize as ``[10, 2]``).

Deserialization (and value-level :meth:`coerce`) is **best-effort**: when a
``target_type`` is supplied and the value cannot be coerced to it, the raw
Expand Down Expand Up @@ -167,6 +167,18 @@ class JsonDataConverter(DataConverter):
> payload; the destination type is always supplied by the caller.
"""

def __init__(
self,
*,
enable_generic_sets: bool = False,
enable_set_coercion: bool = False,
) -> None:
self.enable_generic_sets: bool = enable_generic_sets or enable_set_coercion

@property
def enable_set_coercion(self) -> bool:
return self.enable_generic_sets

def serialize(self, value: Any) -> str | None:
if value is None:
return None
Expand Down Expand Up @@ -221,15 +233,40 @@ def _log_coercion_fallback(target_type: type, error: Exception) -> None:
# ---------------------------------------------------------------------------


def _is_supported_set_element(converter: DataConverter, elem_type: Any) -> bool:
if elem_type is None or elem_type is typing.Any:
return True
elem_origin = typing.get_origin(elem_type)
if elem_origin is not None:
if elem_origin in (list, dict, Sequence, Mapping, set, AbstractSet):
# Mutable collections are unhashable
return False
if elem_origin is typing.Union or elem_origin is types.UnionType:
non_none = [a for a in typing.get_args(elem_type) if a is not type(None)]
return bool(non_none) and all(_is_supported_set_element(converter, a) for a in non_none)
if elem_origin is frozenset:
args = typing.get_args(elem_type)
return not args or _is_supported_set_element(converter, args[0])
return False
if not isinstance(elem_type, type):
return False
if issubclass(elem_type, (list, dict, set)):
return False
return True


def _can_reconstruct(converter: DataConverter, target_type: Any) -> bool:
""":class:`JsonDataConverter`'s reconstruction policy.

Recognizes dataclasses and ``from_json()``-capable types, plus ``Optional``
/ ``list`` / ``Sequence`` hints wrapping them; builtins and unknown
annotations are excluded. Recurses through ``converter.can_reconstruct``
(not itself) so a :class:`JsonDataConverter` subclass that overrides
``can_reconstruct`` still participates in the element-type checks of
``Optional`` / ``list`` hints.
annotations are excluded. When set support is enabled on the converter
(via ``enable_generic_sets=True``), ``set`` / ``frozenset`` / ``Set``
containers with supported element types are also recognized. Recurses
through ``converter.can_reconstruct`` (not itself) so a
:class:`JsonDataConverter` subclass that overrides ``can_reconstruct``
still participates in the element-type checks of ``Optional`` / ``list``
hints.
"""
origin = typing.get_origin(target_type)
if origin is not None:
Expand All @@ -240,11 +277,19 @@ def _can_reconstruct(converter: DataConverter, target_type: Any) -> bool:
)
if origin in (list, Sequence):
return any(converter.can_reconstruct(a) for a in args)
if origin in (set, frozenset, AbstractSet, typing.AbstractSet):
if not getattr(converter, "enable_generic_sets", False):
return False
if not args:
return True
return _is_supported_set_element(converter, args[0])
return False
if not isinstance(target_type, type):
return False
if dataclasses.is_dataclass(target_type):
return True
if target_type in (set, frozenset, AbstractSet):
return getattr(converter, "enable_generic_sets", False)
return callable(getattr(cast(Any, target_type), "from_json", None))


Expand All @@ -253,9 +298,12 @@ def _to_json(obj: Any) -> str:

Builtins serialize to plain JSON. Objects exposing a ``to_json()`` method,
dataclasses, and ``SimpleNamespace`` instances are serialized to plain JSON
as well (without any type marker), recursively. Custom objects can be
reconstructed on the receiving side by passing ``expected_type`` to
:func:`_from_json`.
as well (without any type marker), recursively. Sets and frozensets are
serialized as JSON arrays with elements deterministically ordered by their
canonical JSON string representation (note that numeric set elements follow
lexical JSON ordering, e.g. ``{2, 10}`` serializes as ``[10, 2]``). Custom
objects can be reconstructed on the receiving side by passing ``expected_type``
to :func:`_from_json`.
"""
try:
return json.dumps(obj, default=_encode_custom_object)
Expand Down Expand Up @@ -319,6 +367,31 @@ def _encode_custom_object(o: Any) -> Any:
to_json_hook = getattr(cast(Any, type(o)), "to_json", None)
if callable(to_json_hook):
return to_json_hook(o)
if isinstance(o, (set, frozenset, AbstractSet)):
items: AbstractSet[Any] = cast("AbstractSet[Any]", o)
# Sets and frozensets are unordered collections. To ensure deterministic
# serialization across processes, worker replays, and PYTHONHASHSEED
# variations, elements are individually encoded to canonical JSON
# (with sort_keys=True), sorted lexically by their encoded JSON string,
# and emitted as normalized parsed values.
#
# Emitting the normalized parsed values rather than original objects
# guarantees identical order and eliminates leaks from dict insertion
# order variations, NaN/non-finite comparison anomalies, or partial
# ordering in frozensets. A consequence of sorting canonical JSON strings
# is lexical ordering for numeric sets (e.g., {2, 10} serializes as
# [10, 2]).
try:
encoded_items = sorted(
json.dumps(item, default=_encode_custom_object, sort_keys=True)
for item in items
)
return [json.loads(encoded) for encoded in encoded_items]
except Exception as e:
raise TypeError(
f"Cannot deterministically serialize set of type '{type(items).__name__}' "
f"because its elements cannot be compared or deterministically encoded: {e}"
) from e
if isinstance(o, enum.Enum):
# Emit the member's underlying value (a primitive). Reconstruction is
# type-directed: passing the enum type to ``deserialize`` rebuilds the
Expand Down Expand Up @@ -477,6 +550,19 @@ def _coerce_generic(value: Any, expected_type: Any, origin: Any,
if origin in (list, Sequence) and isinstance(value, list):
elem_type = args[0] if args else None
return [_coerce_to_type(item, elem_type, converter) for item in cast(list[Any], value)]
if origin in (set, frozenset, AbstractSet, typing.AbstractSet) and isinstance(
value, (list, set, frozenset, AbstractSet)
):
if not getattr(converter, "enable_generic_sets", False):
return cast(Any, value)
elem_type = args[0] if args else None
coerced: list[Any] = [
_coerce_to_type(item, elem_type, converter)
for item in cast("Iterable[Any]", value)
]
if origin is frozenset:
return frozenset(coerced)
return set(coerced)
Comment thread
hardikkaurani marked this conversation as resolved.
if origin in (dict, Mapping) and isinstance(value, dict):
# JSON object keys are always strings, so only the values can carry a
# reconstructable type. Keys are passed through unchanged.
Expand Down Expand Up @@ -595,12 +681,12 @@ def _resolve_forward_refs(tp: Any, globalns: dict[str, Any]) -> Any:
"""
if isinstance(tp, str):
try:
tp = eval(tp, globalns) # noqa: S307 - resolves caller-authored annotations
tp = eval(tp, globalns)
except Exception:
return tp
elif isinstance(tp, typing.ForwardRef):
try:
tp = eval(tp.__forward_arg__, globalns) # noqa: S307
tp = eval(tp.__forward_arg__, globalns)
except Exception:
return tp

Expand Down
104 changes: 102 additions & 2 deletions tests/durabletask/test_data_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,14 @@
from unittest.mock import patch

import pytest

from durabletask.internal.entity_state_shim import StateShim
from durabletask.serialization import (
DEFAULT_DATA_CONVERTER,
DataConverter,
JsonDataConverter,
)
from durabletask.internal import type_discovery
from durabletask.worker import _ActivityExecutor, _Registry


@dataclass
Expand Down Expand Up @@ -219,7 +220,7 @@ def __init__(self, /, **kwargs):
self.self = kwargs.get("self", 0)

converter = JsonDataConverter()
encoded = converter.serialize(PositionalReceiver(**{"self": 42}))
encoded = converter.serialize(PositionalReceiver(self=42))
restored = converter.deserialize(encoded, PositionalReceiver)
assert isinstance(restored, PositionalReceiver)
assert restored.self == 42
Expand Down Expand Up @@ -375,3 +376,102 @@ def coerce(self, value: Any, target_type: type | None = None) -> Any:
assert conv.serialize("hello") == '"HELLO"'
assert conv.deserialize('"HELLO"') == "HELLO"
assert conv.coerce("HELLO") == "HELLO"


# ----- set and frozenset container support -----


@dataclass(frozen=True)
class TaggedOrder:
item: str
tags: set[str]
flags: frozenset[int]


def test_roundtrip_dataclass_with_sets():
converter = JsonDataConverter(enable_generic_sets=True)
original = TaggedOrder(item="laptop", tags={"electronics", "sale"}, flags=frozenset({1, 2}))
serialized = converter.serialize(original)
assert serialized is not None
restored = converter.deserialize(serialized, TaggedOrder)
assert isinstance(restored, TaggedOrder)
assert restored.item == "laptop"
assert isinstance(restored.tags, set)
assert restored.tags == {"electronics", "sale"}
assert isinstance(restored.flags, frozenset)
assert restored.flags == frozenset({1, 2})


def test_can_reconstruct_set_and_frozenset():
# By default (enable_generic_sets=False), generic sets are not reconstructed to
# preserve backward compatibility and existing-instance replay behavior.
default_converter = JsonDataConverter()
assert default_converter.can_reconstruct(set[PricedOrder]) is False
assert default_converter.can_reconstruct(frozenset[PricedOrder]) is False
assert default_converter.can_reconstruct(set[int]) is False
assert default_converter.can_reconstruct(set[str]) is False

# When opted in, supported sets (including primitive elements) return True,
# while unsupported unhashable element types return False.
opt_in_converter = JsonDataConverter(enable_generic_sets=True)
assert opt_in_converter.can_reconstruct(set[PricedOrder]) is True
assert opt_in_converter.can_reconstruct(frozenset[PricedOrder]) is True
assert opt_in_converter.can_reconstruct(set[int]) is True
assert opt_in_converter.can_reconstruct(set[str]) is True
assert opt_in_converter.can_reconstruct(frozenset[int]) is True
assert opt_in_converter.can_reconstruct(set[list[str]]) is False
assert opt_in_converter.can_reconstruct(set[dict[str, int]]) is False


def test_existing_instance_replay_compatibility_preserves_list():
"""Preserve existing-instance replay behavior for recorded arrays under set annotations.

When generic set coercion is not explicitly enabled, historical JSON arrays
containing duplicates and order (e.g. {"tags": ["b", "a", "b"]}) remain raw
lists, avoiding NonDeterminismError during orchestration replay.
"""
default_converter = JsonDataConverter()
recorded_input = '{"item": "laptop", "tags": ["b", "a", "b"], "flags": [1, 2, 1]}'
restored = default_converter.deserialize(recorded_input, TaggedOrder)
assert isinstance(restored, TaggedOrder)
assert isinstance(restored.tags, list)
assert restored.tags == ["b", "a", "b"]
assert len(restored.tags) == 3
assert isinstance(restored.flags, list)
assert restored.flags == [1, 2, 1]
assert len(restored.flags) == 3

# With opt-in enabled, set coercion deduplicates and converts to sets
opt_in_converter = JsonDataConverter(enable_generic_sets=True)
opt_in_restored = opt_in_converter.deserialize(recorded_input, TaggedOrder)
assert isinstance(opt_in_restored, TaggedOrder)
assert isinstance(opt_in_restored.tags, set)
assert opt_in_restored.tags == {"a", "b"}
assert len(opt_in_restored.tags) == 2
assert isinstance(opt_in_restored.flags, frozenset)
assert opt_in_restored.flags == frozenset({1, 2})
assert len(opt_in_restored.flags) == 2


def test_type_discovery_and_activity_execution_with_sets():
"""Verify that type discovery recognizes set annotations when opted in,
enabling activities to receive sets rather than raw lists."""
def tag_activity(ctx: Any, tags: set[str]) -> int:
assert isinstance(tags, set)
tags.add("processed")
return len(tags)

reg = _Registry()
reg.add_activity(tag_activity)
logger = logging.getLogger("test_type_discovery_sets")

# With opt-in, activity input is discovered as set[str] and executed
conv = JsonDataConverter(enable_generic_sets=True)
assert type_discovery.activity_input_type(tag_activity, conv) == set[str]
executor = _ActivityExecutor(reg, logger, conv)
output = executor.execute("orch-1", "tag_activity", 1, '["item1", "item2"]')
assert output == "3"

# Default converter preserves input as None (raw JSON), passing list
default_conv = JsonDataConverter()
assert type_discovery.activity_input_type(tag_activity, default_conv) is None
2 changes: 0 additions & 2 deletions tests/durabletask/test_data_converter_roundtrip.py
Original file line number Diff line number Diff line change
Expand Up @@ -545,8 +545,6 @@ class HasUnion:
@pytest.mark.parametrize("value", [
datetime(2020, 1, 1, 12, 0, 0),
Decimal("1.5"),
{1, 2, 3},
frozenset({1, 2}),
b"bytes",
])
def test_unsupported_types_raise_clear_typeerror(conv, value):
Expand Down
Loading
Loading