feat: add envelope serializer for extended python types

Implement EnvelopeSerDes to handle datetime, Decimal, bytes, UUID,
tuple types using wrapper envelope format. Maintains backward
compatibility with existing JSON serializer while providing
comprehensive type support.

- Add TypeHandler chain architecture for extensible serialization
- Support datetime/date with ISO format encoding
- Handle Decimal with string representation
- Support bytes/bytearray/memoryview with base64 encoding
- Add UUID serialization with string format
- Implement tuple/list/dict container handling
- Provide clear error messages for unsupported types
- Add comprehensive test coverage for all supported types

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python/serdes.py

@@ -1,10 +1,19 @@
 """Serialization and deserialization"""
 
+from __future__ import annotations
+
+import base64
 import json
 import logging
+import uuid
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Generic, TypeVar
+from datetime import date, datetime
+from decimal import Decimal
+from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 from aws_durable_execution_sdk_python.exceptions import FatalError
 
@@ -13,6 +22,18 @@
 T = TypeVar("T")
 
 
+class TypeEncoder(Protocol):
+    """Protocol for objects that can encode and decode types."""
+
+    def encode(self, obj: Any) -> dict[str, Any]:
+        """Encode an object to a dictionary representation."""
+        ...
+
+    def decode(self, tag: str, value: Any) -> Any:
+        """Decode a tagged value back to an object."""
+        ...
+
+
 @dataclass(frozen=True)
 class SerDesContext:
     operation_id: str
@@ -37,38 +58,215 @@ def deserialize(self, data: str, _: SerDesContext) -> T:  # noqa: PLR6301
         return json.loads(data)
 
 
+class TypeHandler(TypeEncoder, ABC):
+    def __init__(self, next_handler: TypeEncoder) -> None:
+        self._next: TypeEncoder = next_handler
+
+
+class UnsupportedHandler(TypeEncoder):
+    def encode(self, obj: Any) -> dict[str, Any]:  # noqa: PLR6301
+        msg = f"Unsupported type: {type(obj)!r}"
+        raise TypeError(msg)
+
+    def decode(self, tag: str, value: Any) -> Any:  # noqa: PLR6301, ARG002
+        msg = f"Unknown type tag: {tag!r}"
+        raise ValueError(msg)
+
+
+class BytesHandler(TypeHandler):
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if isinstance(obj, bytes | bytearray | memoryview):
+            encoded: str = base64.b64encode(bytes(obj)).decode()
+            return {"_": {"t": "bytes", "v": encoded}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "bytes":
+            return base64.b64decode(value)
+        return self._next.decode(tag, value)
+
+
+class UuidHandler(TypeHandler):
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if isinstance(obj, uuid.UUID):
+            return {"_": {"t": "uuid", "v": str(obj)}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "uuid":
+            return uuid.UUID(value)
+        return self._next.decode(tag, value)
+
+
+class DecimalHandler(TypeHandler):
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if isinstance(obj, Decimal):
+            return {"_": {"t": "decimal", "v": str(obj)}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "decimal":
+            return Decimal(value)
+        return self._next.decode(tag, value)
+
+
+class DateTimeHandler(TypeHandler):
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if isinstance(obj, datetime):
+            return {"_": {"t": "datetime", "v": obj.isoformat()}}
+        if isinstance(obj, date):
+            return {"_": {"t": "date", "v": obj.isoformat()}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "datetime":
+            return datetime.fromisoformat(value)
+        if tag == "date":
+            return date.fromisoformat(value)
+        return self._next.decode(tag, value)
+
+
+class ContainerHandler(TypeHandler):
+    def __init__(self, next_handler: TypeEncoder) -> None:
+        super().__init__(next_handler)
+        self._dispatch_encode: Callable[[Any], dict[str, Any]] | None = None
+        self._dispatch_decode: Callable[[str, Any], Any] | None = None
+
+    def _enc(self, obj: Any) -> dict[str, Any]:
+        if self._dispatch_encode is None:
+            msg = "ContainerHandler not initialized with encode dispatcher."
+            raise RuntimeError(msg)
+        return self._dispatch_encode(obj)
+
+    def _dec(self, tag: str, value: Any) -> Any:
+        if self._dispatch_decode is None:
+            msg = "ContainerHandler not initialized with decode dispatcher."
+            raise RuntimeError(msg)
+        return self._dispatch_decode(tag, value)
+
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if isinstance(obj, tuple):
+            items: list[dict[str, Any]] = [self._enc(x) for x in obj]
+            return {"_": {"t": "tuple", "v": items}}
+        if isinstance(obj, list):
+            items_list: list[dict[str, Any]] = [self._enc(x) for x in obj]
+            return {"_": {"t": "list", "v": items_list}}
+        if isinstance(obj, dict):
+            self._validate_dict_keys(obj)
+            wrapped: dict[str, dict[str, Any]] = {
+                k: self._enc(v) for k, v in obj.items()
+            }
+            return {"_": {"t": "dict", "v": wrapped}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "tuple":
+            if not isinstance(value, list):
+                msg = 'Malformed envelope: "tuple" expects array value.'
+                raise TypeError(msg)
+            return tuple(self._dec(v["_"]["t"], v["_"]["v"]) for v in value)
+        if tag == "list":
+            if not isinstance(value, list):
+                msg = 'Malformed envelope: "list" expects array value.'
+                raise TypeError(msg)
+            return [self._dec(v["_"]["t"], v["_"]["v"]) for v in value]
+        if tag == "dict":
+            if not isinstance(value, dict):
+                msg = 'Malformed envelope: "dict" expects object value.'
+                raise TypeError(msg)
+            return {k: self._dec(v["_"]["t"], v["_"]["v"]) for k, v in value.items()}
+        return self._next.decode(tag, value)
+
+    @staticmethod
+    def _validate_dict_keys(mapping: dict[Any, Any]) -> None:
+        bad: list[Any] = [k for k in mapping if not isinstance(k, str)]
+        if bad:
+            ex: Any = bad[0]
+            msg = f"Unsupported mapping key type: {type(ex)!r}. JSON object keys must be strings."
+            raise TypeError(msg)
+
+
+class PrimitiveHandler(TypeHandler):
+    def encode(self, obj: Any) -> dict[str, Any]:
+        if obj is None or isinstance(obj, str | int | float | bool):
+            tag: str = type(obj).__name__
+            return {"_": {"t": tag, "v": obj}}
+        return self._next.encode(obj)
+
+    def decode(self, tag: str, value: Any) -> Any:
+        if tag == "NoneType":
+            return None
+        if tag in {"str", "int", "float", "bool"}:
+            return value
+        return self._next.decode(tag, value)
+
+
+@dataclass(frozen=True)
+class HandlerChain:
+    root: TypeHandler
+    container: ContainerHandler
+
+    @classmethod
+    def create(cls) -> HandlerChain:
+        unsupported: UnsupportedHandler = UnsupportedHandler()
+        bytes_h: BytesHandler = BytesHandler(unsupported)
+        uuid_h: UuidHandler = UuidHandler(bytes_h)
+        decimal_h: DecimalHandler = DecimalHandler(uuid_h)
+        dt_h: DateTimeHandler = DateTimeHandler(decimal_h)
+        container_h: ContainerHandler = ContainerHandler(dt_h)
+        primitive_h: PrimitiveHandler = PrimitiveHandler(container_h)
+
+        # Wire dispatchers to always go through the root
+        container_h._dispatch_encode = primitive_h.encode  # noqa: SLF001
+        container_h._dispatch_decode = primitive_h.decode  # noqa: SLF001
+
+        return cls(root=primitive_h, container=container_h)
+
+
+class EnvelopeSerDes(SerDes[T]):
+    def __init__(self) -> None:
+        self._chain: HandlerChain = HandlerChain.create()
+
+    def serialize(self, value: T, _: SerDesContext) -> str:
+        wrapped: dict[str, Any] = self._chain.root.encode(value)
+        return json.dumps(wrapped, separators=(",", ":"))
+
+    def deserialize(self, data: str, _: SerDesContext) -> T:
+        obj: Any = json.loads(data)
+        if not (isinstance(obj, dict) and "_" in obj and isinstance(obj["_"], dict)):
+            msg = 'Malformed envelope: root must be {"_": {"t": ..., "v": ...}}.'
+            raise TypeError(msg)
+        inner: dict[str, Any] = obj["_"]
+        if not (isinstance(inner, dict) and "t" in inner and "v" in inner):
+            msg = 'Malformed envelope: missing "t" or "v" at root.'
+            raise TypeError(msg)
+        return self._chain.root.decode(inner["t"], inner["v"])
+
+
 _DEFAULT_JSON_SERDES: SerDes = JsonSerDes()
 
 
 def serialize(
     serdes: SerDes[T] | None, value: T, operation_id: str, durable_execution_arn: str
 ) -> str:
     serdes_context: SerDesContext = SerDesContext(operation_id, durable_execution_arn)
-    if serdes is None:
-        serdes = _DEFAULT_JSON_SERDES
+    active_serdes: SerDes[T] = serdes or _DEFAULT_JSON_SERDES
     try:
-        return serdes.serialize(value, serdes_context)
+        return active_serdes.serialize(value, serdes_context)
     except Exception as e:
-        logger.exception(
-            "⚠️ Serialization failed for id: %s",
-            operation_id,
-        )
-        msg = f"Serialization failed for id: {operation_id}, error: {e}."
+        logger.exception("⚠️ Serialization failed for id: %s", operation_id)
+        msg: str = f"Serialization failed for id: {operation_id}, error: {e}."
         raise FatalError(msg) from e
 
 
 def deserialize(
     serdes: SerDes[T] | None, data: str, operation_id: str, durable_execution_arn: str
 ) -> T:
     serdes_context: SerDesContext = SerDesContext(operation_id, durable_execution_arn)
-    if serdes is None:
-        serdes = _DEFAULT_JSON_SERDES
+    active_serdes: SerDes[T] = serdes or _DEFAULT_JSON_SERDES
     try:
-        return serdes.deserialize(data, serdes_context)
+        return active_serdes.deserialize(data, serdes_context)
     except Exception as e:
-        logger.exception(
-            "⚠️ Deserialization failed for id: %s",
-            operation_id,
-        )
-        msg = f"Deserialization failed for id: {operation_id}"
+        logger.exception("⚠️ Deserialization failed for id: %s", operation_id)
+        msg: str = f"Deserialization failed for id: {operation_id}"
         raise FatalError(msg) from e