Add pydantic example, fix reconstructibility concern

Author: andystaples

CHANGELOG.md

@@ -46,6 +46,14 @@ ADDED
   `DataConverter` as a second parameter (`from_json(cls, value, converter)`),
   letting it reconstruct nested typed values via `converter.coerce(...)` /
   `converter.deserialize(...)`. The single-argument form remains supported.
+- `DataConverter` now exposes an overridable `is_reconstructable(target_type)`
+  method that controls which annotated input/return types the SDK reconstructs
+  on the inbound path. A custom converter can override it to recognize its own
+  types (for example `pydantic.BaseModel` subclasses), so that orchestrator /
+  activity / entity inputs annotated with those types are reconstructed by the
+  converter instead of arriving as raw JSON. The default behavior is unchanged
+  (dataclasses and `from_json()`-capable types, plus `Optional` / `list`
+  wrappers, are reconstructable; builtins are not).
 - Added `EntityMetadata.get_typed_state(intended_type=...)`, which deserializes
   the entity's persisted state and reconstructs dataclasses and
   `from_json()`-capable types. The existing `get_state()` is unchanged: with no

durabletask/internal/type_discovery.py

@@ -5,54 +5,36 @@
 
 These helpers resolve the annotation of the *input* parameter of an
 orchestrator, activity, or entity function so that inbound payloads can be
-reconstructed into the annotated custom type (a dataclass or a type exposing a
-``from_json()`` classmethod) without the caller having to pass an explicit type.
+reconstructed into the annotated custom type without the caller having to pass
+an explicit type.
 
 Discovery is intentionally conservative: it only returns an annotation when the
-target is a *reconstructable* custom type (a dataclass, a ``from_json()``-capable
-type, or an ``Optional`` / ``list`` wrapping one). Primitive and unknown
-annotations resolve to ``None`` so that existing payloads are passed through
-unchanged -- inbound type discovery never invokes an arbitrary constructor on
-untrusted data, and never alters the value for builtins.
+active :class:`~durabletask.serialization.DataConverter` reports it as
+*reconstructable* via :meth:`DataConverter.is_reconstructable`. The default
+converter recognizes dataclasses, ``from_json()``-capable types, and ``Optional``
+/ ``list`` hints wrapping them; a custom converter can recognize its own types
+(e.g. ``pydantic.BaseModel``). Primitive and unknown annotations resolve to
+``None`` so that existing payloads are passed through unchanged -- inbound type
+discovery never invokes an arbitrary constructor on untrusted data, and never
+alters the value for builtins.
 
 All public helpers swallow exceptions and return ``None`` on failure; the caller
 treats ``None`` as "no type information available" and uses the raw payload.
 """
 
 from __future__ import annotations
 
-import collections.abc
-import dataclasses
 import functools
 import inspect
-import types
 import typing
-from typing import Any, Callable, cast
+from typing import Any, Callable
 
+from durabletask.serialization import DEFAULT_DATA_CONVERTER, DataConverter
 
-def is_reconstructable(annotation: Any) -> bool:
-    """Return True if ``annotation`` names a custom type we can rebuild.
 
-    Reconstructable targets are dataclasses, types exposing a callable
-    ``from_json``, and ``Optional`` / ``list`` hints wrapping such types.
-    Builtins (``int``, ``str``, ``dict``, ...) and unknown annotations are not
-    reconstructable and resolve to ``False``.
-    """
-    origin = typing.get_origin(annotation)
-    if origin is not None:
-        args = typing.get_args(annotation)
-        if origin is typing.Union or origin is types.UnionType:
-            return any(
-                is_reconstructable(a) for a in args if a is not type(None)
-            )
-        if origin in (list, collections.abc.Sequence):
-            return any(is_reconstructable(a) for a in args)
-        return False
-    if not isinstance(annotation, type):
-        return False
-    if dataclasses.is_dataclass(annotation):
-        return True
-    return callable(getattr(cast(Any, annotation), "from_json", None))
+def _resolve_converter(converter: DataConverter | None) -> DataConverter:
+    """Return the supplied converter, or the shared default when ``None``."""
+    return converter if converter is not None else DEFAULT_DATA_CONVERTER
 
 
 # Bounded so a worker that registers dynamically-created functions or closures
@@ -72,14 +54,15 @@ def _resolved_hints(fn: Callable[..., Any]) -> dict[str, Any] | None:
         return None
 
 
-def _input_annotation(fn: Callable[..., Any], position: int) -> Any | None:
+def _input_annotation(fn: Callable[..., Any], position: int,
+                      converter: DataConverter | None = None) -> Any | None:
     """Return the resolved annotation of the positional parameter at ``position``.
 
     ``position`` is the zero-based index among positional parameters (so the
     ``input`` parameter of a ``(ctx, input)`` function is at position 1, and the
     ``input`` parameter of an unbound ``(self, input)`` entity method is also at
     position 1). Returns ``None`` when the parameter is absent, unannotated, or
-    its annotation is not a reconstructable custom type.
+    its annotation is not reconstructable by ``converter``.
     """
     try:
         sig = inspect.signature(fn)
@@ -105,27 +88,29 @@ def _input_annotation(fn: Callable[..., Any], position: int) -> Any | None:
 
     if annotation is inspect.Parameter.empty or annotation is Any:
         return None
-    return annotation if is_reconstructable(annotation) else None
+    return annotation if _resolve_converter(converter).is_reconstructable(annotation) else None
 
 
-def orchestrator_input_type(fn: Callable[..., Any]) -> Any | None:
+def orchestrator_input_type(fn: Callable[..., Any],
+                            converter: DataConverter | None = None) -> Any | None:
     """Discover the input type of an orchestrator function ``(ctx, input)``."""
-    return _input_annotation(fn, 1)
+    return _input_annotation(fn, 1, converter)
 
 
-def activity_input_type(fn: Callable[..., Any]) -> Any | None:
+def activity_input_type(fn: Callable[..., Any],
+                        converter: DataConverter | None = None) -> Any | None:
     """Discover the input type of an activity function ``(ctx, input)``."""
-    return _input_annotation(fn, 1)
+    return _input_annotation(fn, 1, converter)
 
 
-def activity_output_type(fn: Any) -> Any | None:
+def activity_output_type(fn: Any, converter: DataConverter | None = None) -> Any | None:
     """Discover the return type of an activity function.
 
-    Returns the resolved return annotation when it names a reconstructable
-    custom type (a dataclass or a ``from_json()``-capable type, optionally
-    wrapped in ``Optional`` / ``list``). Returns ``None`` for plain callables
-    that are not annotated with such a type, for string activity names, or when
-    the annotation cannot be resolved.
+    Returns the resolved return annotation when ``converter`` reports it as
+    reconstructable (the default converter recognizes a dataclass or a
+    ``from_json()``-capable type, optionally wrapped in ``Optional`` / ``list``).
+    Returns ``None`` for plain callables that are not annotated with such a type,
+    for string activity names, or when the annotation cannot be resolved.
     """
     if not callable(fn):
         return None
@@ -144,10 +129,11 @@ def activity_output_type(fn: Any) -> Any | None:
 
     if annotation is inspect.Signature.empty or annotation is Any or annotation is None:
         return None
-    return annotation if is_reconstructable(annotation) else None
+    return annotation if _resolve_converter(converter).is_reconstructable(annotation) else None
 
 
-def entity_input_type(fn: Any, operation: str) -> Any | None:
+def entity_input_type(fn: Any, operation: str,
+                      converter: DataConverter | None = None) -> Any | None:
     """Discover the input type of an entity operation.
 
     For class-based entities (a ``DurableEntity`` subclass) the operation is a
@@ -160,5 +146,5 @@ def entity_input_type(fn: Any, operation: str) -> Any | None:
         if method is None or not callable(method):
             return None
         # Unbound method includes ``self`` at position 0, so ``input`` is at 1.
-        return _input_annotation(method, 1)
-    return _input_annotation(fn, 1)
+        return _input_annotation(method, 1, converter)
+    return _input_annotation(fn, 1, converter)

durabletask/serialization.py

@@ -101,6 +101,30 @@ def coerce(self, value: Any, target_type: type | None = None) -> Any:
         """
         ...
 
+    def is_reconstructable(self, target_type: Any) -> bool:
+        """Return True if this converter can rebuild ``target_type`` from a payload.
+
+        Inbound type-discovery calls this to decide whether a function's
+        annotated *input* type (or an activity's *return* annotation) should be
+        passed to :meth:`deserialize` / :meth:`coerce`. When it returns ``False``
+        the SDK passes the raw deserialized payload through unchanged -- this
+        gate is what stops the SDK from invoking an arbitrary constructor on a
+        builtin or otherwise unrecognized annotation.
+
+        The default recognizes the types the built-in codec can rebuild --
+        dataclasses and ``from_json()``-capable types, plus ``Optional`` /
+        ``list`` / ``Sequence`` hints wrapping them -- and excludes builtins
+        (``int``, ``str``, ``dict``, ...) and unknown annotations.
+
+        Override this to teach the SDK about a custom converter's own types (for
+        example ``pydantic.BaseModel`` subclasses) so that inputs annotated with
+        them are reconstructed instead of arriving as raw JSON. The default
+        implementation recurses through ``self.is_reconstructable``, so an
+        override is also consulted for the element types of ``Optional`` /
+        ``list`` hints (e.g. ``list[MyModel]``).
+        """
+        return _is_reconstructable(self, target_type)
+
 
 class JsonDataConverter(DataConverter):
     """Default :class:`DataConverter` backed by the SDK's JSON codec.
@@ -187,6 +211,32 @@ def _log_coercion_fallback(target_type: type, error: Exception) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _is_reconstructable(converter: DataConverter, target_type: Any) -> bool:
+    """Default :meth:`DataConverter.is_reconstructable` policy.
+
+    Recognizes dataclasses and ``from_json()``-capable types, plus ``Optional``
+    / ``list`` / ``Sequence`` hints wrapping them; builtins and unknown
+    annotations are excluded. Recurses through ``converter.is_reconstructable``
+    (not itself) so a subclass override participates in the element-type checks
+    of ``Optional`` / ``list`` hints.
+    """
+    origin = typing.get_origin(target_type)
+    if origin is not None:
+        args = typing.get_args(target_type)
+        if origin is typing.Union or origin is types.UnionType:
+            return any(
+                converter.is_reconstructable(a) for a in args if a is not type(None)
+            )
+        if origin in (list, Sequence):
+            return any(converter.is_reconstructable(a) for a in args)
+        return False
+    if not isinstance(target_type, type):
+        return False
+    if dataclasses.is_dataclass(target_type):
+        return True
+    return callable(getattr(cast(Any, target_type), "from_json", None))
+
+
 def _to_json(obj: Any) -> str:
     """Serialize a value to a JSON string.
 

durabletask/worker.py

@@ -1692,7 +1692,7 @@ def call_activity(
         # converter decides how a coercion failure is handled (the default is
         # best-effort).
         if return_type is None and not isinstance(activity, str):
-            return_type = type_discovery.activity_output_type(activity)
+            return_type = type_discovery.activity_output_type(activity, self._data_converter)
 
         self.call_activity_function_helper(
             id, activity, input=input, retry_policy=retry_policy, is_sub_orch=False, tags=tags,
@@ -2211,7 +2211,7 @@ def process_event(
                 if (
                         event.executionStarted.HasField("input") and event.executionStarted.input.value != ""
                 ):
-                    input_type = type_discovery.orchestrator_input_type(fn)
+                    input_type = type_discovery.orchestrator_input_type(fn, self._data_converter)
                     input = self._data_converter.deserialize(
                         event.executionStarted.input.value, input_type)
 
@@ -2856,7 +2856,7 @@ def execute(
                 f"Activity function named '{name}' was not registered!"
             )
 
-        input_type = type_discovery.activity_input_type(fn) if encoded_input else None
+        input_type = type_discovery.activity_input_type(fn, self._data_converter) if encoded_input else None
         activity_input = self._data_converter.deserialize(encoded_input, input_type)
         ctx = task.ActivityContext(orchestration_id, task_id)
 
@@ -2897,7 +2897,7 @@ def execute(
                 f"Entity function named '{entity_id.entity}' was not registered!"
             )
 
-        input_type = type_discovery.entity_input_type(fn, operation) if encoded_input else None
+        input_type = type_discovery.entity_input_type(fn, operation, self._data_converter) if encoded_input else None
         entity_input = self._data_converter.deserialize(encoded_input, input_type)
         ctx = EntityContext(orchestration_id, operation, state, entity_id, self._data_converter)
 

examples/README.md

@@ -169,6 +169,12 @@ python activity_sequence.py
 > account and an additional install step. See the
 > [large payload README](./large_payload/README.md) for details.
 
+> [!NOTE]
+> The `custom_data_converter/` example is a self-contained subproject that
+> shows how to plug a third-party serialization library (pydantic) into the
+> SDK via a custom `DataConverter`, with in-process tests that need no backend.
+> See the [custom DataConverter README](./custom_data_converter/README.md).
+
 ### Review Orchestration History and Status
 
 To access the Durable Task Scheduler Dashboard, follow these steps: