Add explicit Avro payload adapters

Issue: zorporation/durable-workflow#538

Loop-ID: build-02

Author: durable-workflow-ops

README.md

@@ -198,6 +198,53 @@ scalars before passing them to the SDK. `IntEnum` and `StrEnum` encode because
 they are JSON scalar subclasses, but they decode as `int` and `str`.
 `OrderedDict` decodes as a plain `dict`.
 
+Use `to_avro_payload_value(...)` when a rich value should enter durable
+history through the default Avro envelope:
+
+```python
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from decimal import Decimal
+from enum import Enum
+from uuid import UUID
+
+from durable_workflow import Client, to_avro_payload_value
+
+
+class OrderStatus(Enum):
+    PENDING = "pending"
+
+
+@dataclass
+class OrderInput:
+    order_id: UUID
+    placed_at: datetime
+    amount: Decimal
+    status: OrderStatus
+
+
+order = OrderInput(
+    order_id=UUID("12345678-1234-5678-1234-567812345678"),
+    placed_at=datetime.now(timezone.utc),
+    amount=Decimal("10.25"),
+    status=OrderStatus.PENDING,
+)
+
+client = Client("http://server:8080", token="dev-token-123")
+await client.start_workflow(
+    "order-workflow",
+    task_queue="orders",
+    workflow_id="order-123",
+    input=[to_avro_payload_value(order)],
+)
+```
+
+The helper also accepts pydantic-style models with `model_dump(mode="json")`
+and attrs-style classes. Rebuild domain objects explicitly inside workflows or
+activities, for example `OrderInput(order_id=UUID(data["order_id"]), ...)`.
+Adapter output is part of the durable history contract, so changing that shape
+is a workflow compatibility change.
+
 ## Authentication
 
 For local servers that use one shared bearer token, pass `token=`:

src/durable_workflow/__init__.py

@@ -58,7 +58,12 @@
     PrometheusMetrics,
 )
 from .retry_policy import RetryPolicy, TransportRetryPolicy
-from .serializer import PayloadSizeWarningConfig, PayloadSizeWarningContext
+from .serializer import (
+    PayloadSizeWarningConfig,
+    PayloadSizeWarningContext,
+    to_avro_payload_value,
+    to_avro_payload_values,
+)
 from .worker import Worker
 from .workflow import ActivityRetryPolicy, ChildWorkflowRetryPolicy, ContinueAsNew, StartChildWorkflow
 
@@ -120,4 +125,6 @@
     "WorkflowFailed",
     "WorkflowNotFound",
     "WorkflowTerminated",
+    "to_avro_payload_value",
+    "to_avro_payload_values",
 ]

src/durable_workflow/serializer.py

@@ -21,8 +21,12 @@
 import json
 import logging
 from collections.abc import Mapping, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, fields, is_dataclass
+from datetime import date, datetime, time
+from decimal import Decimal
+from enum import Enum
 from typing import Any, TypeGuard, cast
+from uuid import UUID
 
 from . import _avro
 
@@ -91,6 +95,94 @@ def to_log_context(self) -> dict[str, str]:
 PayloadWarningContexts = PayloadWarningContext | Sequence[PayloadWarningContext]
 
 
+def to_avro_payload_value(value: Any) -> Any:
+    """Convert common rich Python values to JSON-native Avro wrapper values.
+
+    The SDK's default Avro codec is a language-neutral envelope around a JSON
+    document. This helper is the explicit boundary for class-carrying values:
+    callers opt in before encode and the returned value becomes part of durable
+    history.
+    """
+    if isinstance(value, Enum):
+        return to_avro_payload_value(value.value)
+
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+
+    if isinstance(value, datetime):
+        return value.isoformat()
+
+    if isinstance(value, date | time):
+        return value.isoformat()
+
+    if isinstance(value, UUID | Decimal):
+        return str(value)
+
+    if isinstance(value, Mapping):
+        converted: dict[str, Any] = {}
+        for key, item in value.items():
+            if not isinstance(key, str):
+                raise TypeError("Avro JSON payload dictionaries must use string keys after adaptation")
+            converted[key] = to_avro_payload_value(item)
+        return converted
+
+    if is_dataclass(value) and not isinstance(value, type):
+        return {
+            field.name: to_avro_payload_value(getattr(value, field.name))
+            for field in fields(value)
+        }
+
+    pydantic_dump = _pydantic_model_dump(value)
+    if pydantic_dump is not None:
+        return to_avro_payload_value(pydantic_dump)
+
+    attrs_dump = _attrs_payload_dict(value)
+    if attrs_dump is not None:
+        return attrs_dump
+
+    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
+        return [to_avro_payload_value(item) for item in value]
+
+    raise TypeError(
+        f"Object of type {type(value).__name__} is not Avro JSON payload safe; "
+        "adapt it to None, bool, int, float, str, list, or dict[str, value] first."
+    )
+
+
+def to_avro_payload_values(values: Sequence[Any]) -> list[Any]:
+    """Convert several values with :func:`to_avro_payload_value`."""
+    return [to_avro_payload_value(value) for value in values]
+
+
+def _pydantic_model_dump(value: Any) -> Any | None:
+    if not (
+        hasattr(value, "__pydantic_fields__")
+        or hasattr(value, "__fields__")
+        or value.__class__.__module__.startswith("pydantic.")
+    ):
+        return None
+
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        return model_dump(mode="json")
+
+    dict_dump = getattr(value, "dict", None)
+    if callable(dict_dump):
+        return dict_dump()
+
+    return None
+
+
+def _attrs_payload_dict(value: Any) -> dict[str, Any] | None:
+    attrs_fields = getattr(value.__class__, "__attrs_attrs__", None)
+    if attrs_fields is None:
+        return None
+    return {
+        field.name: to_avro_payload_value(getattr(value, field.name))
+        for field in attrs_fields
+    }
+
+
 def encode(
     value: Any,
     codec: str = AVRO_CODEC,

tests/test_serializer.py

@@ -34,6 +34,14 @@ class SerializerDataclass:
     count: int
 
 
+@dataclass
+class SerializerOrder:
+    order_id: UUID
+    placed_at: datetime
+    amount: Decimal
+    status: "SerializerEnum"
+
+
 class SerializerEnum(Enum):
     PENDING = "pending"
 
@@ -50,6 +58,32 @@ class SerializerStrEnum(StrEnum):
     SerializerStrEnum = None
 
 
+class _AttrsField:
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+
+class SerializerAttrsStyle:
+    __attrs_attrs__ = (_AttrsField("sku"), _AttrsField("quantity"))
+
+    def __init__(self, sku: str, quantity: int) -> None:
+        self.sku = sku
+        self.quantity = quantity
+
+
+class SerializerPydanticStyle:
+    __pydantic_fields__ = {"order_id": object()}
+
+    def __init__(self, order_id: UUID, due_on: date) -> None:
+        self.order_id = order_id
+        self.due_on = due_on
+
+    def model_dump(self, *, mode: str = "python") -> dict[str, object]:
+        if mode == "json":
+            return {"order_id": str(self.order_id), "due_on": self.due_on.isoformat()}
+        return {"order_id": self.order_id, "due_on": self.due_on}
+
+
 class TestEncode:
     def test_list(self) -> None:
         assert serializer.encode(["a", 1, True], codec="json") == '["a",1,true]'
@@ -166,6 +200,66 @@ def test_encode_many_rejects_context_count_mismatch(self) -> None:
             )
 
 
+class TestAvroPayloadAdapter:
+    def test_adapts_dataclass_datetime_uuid_decimal_and_enum(self) -> None:
+        order = SerializerOrder(
+            order_id=UUID("12345678-1234-5678-1234-567812345678"),
+            placed_at=datetime(2026, 4, 21, 10, 30, tzinfo=timezone.utc),
+            amount=Decimal("10.25"),
+            status=SerializerEnum.PENDING,
+        )
+
+        assert serializer.to_avro_payload_value(order) == {
+            "order_id": "12345678-1234-5678-1234-567812345678",
+            "placed_at": "2026-04-21T10:30:00+00:00",
+            "amount": "10.25",
+            "status": "pending",
+        }
+
+    def test_adapts_pydantic_style_models_through_json_mode_dump(self) -> None:
+        model = SerializerPydanticStyle(
+            UUID("12345678-1234-5678-1234-567812345678"),
+            date(2026, 4, 21),
+        )
+
+        assert serializer.to_avro_payload_value(model) == {
+            "order_id": "12345678-1234-5678-1234-567812345678",
+            "due_on": "2026-04-21",
+        }
+
+    def test_adapts_attrs_style_objects_and_sequences(self) -> None:
+        value = (SerializerAttrsStyle("ABC", 2), time(10, 30, tzinfo=timezone.utc))
+
+        assert serializer.to_avro_payload_value(value) == [
+            {"sku": "ABC", "quantity": 2},
+            "10:30:00+00:00",
+        ]
+
+    @requires_avro
+    def test_adapter_output_round_trips_through_default_avro_codec(self) -> None:
+        value = serializer.to_avro_payload_value(
+            {
+                "model": SerializerDataclass(name="Ada", count=2),
+                "ids": [UUID("12345678-1234-5678-1234-567812345678")],
+            }
+        )
+
+        blob = serializer.encode(value, codec="avro")
+
+        assert serializer.decode(blob, codec="avro") == {
+            "model": {"name": "Ada", "count": 2},
+            "ids": ["12345678-1234-5678-1234-567812345678"],
+        }
+
+    def test_adapter_rejects_non_string_mapping_keys(self) -> None:
+        with pytest.raises(TypeError, match="string keys"):
+            serializer.to_avro_payload_value({1: "one"})
+
+    def test_adapter_rejects_unadapted_objects(self) -> None:
+        with pytest.raises(TypeError, match="not Avro JSON payload safe"):
+            serializer.to_avro_payload_value(object())
+
+
 class TestPayloadSizeWarning:
     def test_encode_warns_with_structured_context(self, caplog: pytest.LogCaptureFixture) -> None:
         config = serializer.PayloadSizeWarningConfig(limit_bytes=10, threshold_percent=50)