feat: remove pydantic and allow custom type codec

Author: brian14708

.github/workflows/ci.yaml

@@ -27,7 +27,7 @@ jobs:
         with:
           enable-cache: true
       - name: Sync dependencies
-        run: uv sync --group type-checking --group lint --extra full
+        run: uv sync --group type-checking --group lint
       - name: Run tests
         run: uv run pytest
       - name: Run linter

noxfile.py

@@ -12,7 +12,7 @@
 
 def install_deps(s: nox.Session, groups: list[str]):
     s.env["UV_PROJECT_ENVIRONMENT"] = s.virtualenv.location
-    cmd = ["uv", "sync", "--frozen", "--extra", "full"]
+    cmd = ["uv", "sync", "--frozen"]
     for g in groups:
         cmd.extend(("--group", g))
     _ = s.run_install(*cmd)

pyproject.toml

@@ -8,11 +8,6 @@ dependencies = [
     "typing-extensions>=4.15.0",
 ]
 
-[project.optional-dependencies]
-full = [
-    "pydantic>=2.11.9",
-]
-
 [build-system]
 requires = ["uv_build>=0.8.19,<0.9.0"]
 build-backend = "uv_build"
@@ -25,7 +20,6 @@ dev = [
 type-checking = [
     "basedmypy>=2.10.0",
     "basedpyright>=1.31.2",
-    "pydantic>=2.11.9",
 ]
 lint = [
     "ruff>=0.12.8",

src/duron/codec.py

@@ -1,85 +1,30 @@
 from __future__ import annotations
 
-import base64
-import importlib
-import pickle
-from typing import TYPE_CHECKING, cast, final
+from typing import TYPE_CHECKING, final
 
 from typing_extensions import Protocol, override
 
 from duron.log import is_json_value
 
-BaseModel: type[_pydantic.BaseModel] | None = None
-try:
-    import pydantic as _pydantic
-
-    BaseModel = _pydantic.BaseModel
-except ImportError:
-    pass
-
 if TYPE_CHECKING:
     from duron.log import JSONValue
 
-__all__ = ["Codec", "DEFAULT_CODEC"]
+__all__ = ["Codec", "DefaultCodec"]
 
 
 class Codec(Protocol):
     def encode_json(self, result: object) -> JSONValue: ...
     def decode_json(self, encoded: JSONValue) -> object: ...
 
-    def encode_state(self, obj: object) -> str: ...
-    def decode_state(self, state: str) -> object: ...
-
 
 @final
-class _DefaultCodec(Codec):
-    def __init__(self) -> None:
-        self._type_cache: dict[str, type] = {}
-
-    def _lookup_model(self, qual: str) -> type:
-        cls = self._type_cache.get(qual)
-        if cls is not None:
-            return cls
-
-        mod, name = qual.split(":", 1)
-        obj: object = importlib.import_module(mod)
-        for part in name.split("."):
-            obj = getattr(obj, part)
-        if isinstance(obj, type):
-            self._type_cache[qual] = obj
-            return obj
-        else:
-            raise TypeError(f"Imported object is not a type: {obj!r}")
-
+class DefaultCodec(Codec):
     @override
     def encode_json(self, result: object) -> JSONValue:
-        if BaseModel and isinstance(result, BaseModel):
-            cls = result.__class__
-            model = result.model_dump(mode="json")
-            model["_duron.pydantic"] = f"{cls.__module__}:{cls.__qualname__}"
-            return model
         if is_json_value(result):
             return result
         raise TypeError(f"Result is not JSON-serializable: {result!r}")
 
     @override
     def decode_json(self, encoded: JSONValue) -> object:
-        if isinstance(encoded, dict) and "_duron.pydantic" in encoded:
-            model = self._lookup_model(cast("str", encoded["_duron.pydantic"]))
-            if not BaseModel or not issubclass(model, BaseModel):
-                raise TypeError(f"Decoded class is not a BaseModel subclass: {model}")
-            return model.model_validate({
-                k: v for k, v in encoded.items() if not k.startswith("_duron.")
-            })
         return encoded
-
-    @override
-    def encode_state(self, obj: object) -> str:
-        return base64.b64encode(pickle.dumps(obj, protocol=5)).decode("ascii")
-
-    @override
-    def decode_state(self, state: str) -> object:
-        return pickle.loads(base64.b64decode(state.encode()))
-
-
-DEFAULT_CODEC: Codec = _DefaultCodec()

src/duron/fn.py

@@ -8,7 +8,7 @@
     TypeVar,
 )
 
-from duron.codec import DEFAULT_CODEC
+from duron.codec import DefaultCodec
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -30,14 +30,14 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _T_co:
 
 
 def durable(
-    *, codec: Codec = DEFAULT_CODEC
+    *, codec: Codec | None = None
 ) -> Callable[[Callable[_P, _T_co]], DurableFn[_P, _T_co]]:
     """
     Mark a function as durable, meaning its execution can be recorded and
     replayed.
     """
 
     def decorate(fn: Callable[_P, _T_co]) -> DurableFn[_P, _T_co]:
-        return DurableFn(codec=codec, fn=fn)
+        return DurableFn(codec=codec or DefaultCodec(), fn=fn)
 
     return decorate

src/duron/log/__init__.py

@@ -27,7 +27,6 @@ class _BaseEntry(TypedDict):
 class ErrorInfo(TypedDict):
     code: int
     message: str
-    state: NotRequired[str]  # opaque
 
 
 class PromiseCreateEntry(_BaseEntry):
@@ -49,7 +48,7 @@ class StreamEmitEntry(_BaseEntry):
     type: Literal["stream/emit"]
     stream_id: str
     value: NotRequired[JSONValue]
-    state: NotRequired[str]  # opaque
+    state: NotRequired[JSONValue]
 
 
 class StreamCompleteEntry(_BaseEntry):

src/duron/task.py

@@ -242,7 +242,7 @@ async def handle_message(self, e: Entry) -> None:
             if "error" in e:
                 self._loop.post_completion_threadsafe(
                     id,
-                    exception=_decode_error(e["error"], self._codec),
+                    exception=_decode_error(e["error"]),
                 )
                 self._pending_ops.discard(id)
             elif "result" in e:
@@ -287,7 +287,7 @@ async def cb() -> None:
                             result = await cast("Awaitable[object]", result)
                         entry["result"] = self._codec.encode_json(result)
                     except BaseException as e:
-                        entry["error"] = _encode_error(e, self._codec)
+                        entry["error"] = _encode_error(e)
                     await self.enqueue_log(entry)
 
                 if self._running:
@@ -313,7 +313,7 @@ async def cb() -> None:
                         result = await op.task
                         entry["result"] = self._codec.encode_json(result)
                     except BaseException as e:
-                        entry["error"] = _encode_error(e, self._codec)
+                        entry["error"] = _encode_error(e)
 
                     await self.enqueue_log(entry, True)
 
@@ -343,20 +343,12 @@ def _decode_timestamp(ts: int) -> int:
     return ts * 1_000
 
 
-def _encode_error(error: BaseException, codec: Codec) -> ErrorInfo:
-    """Convert exception to ErrorInfo dict."""
+def _encode_error(error: BaseException) -> ErrorInfo:
     return {
         "code": -1,
         "message": str(error),
-        "state": codec.encode_state(error),
     }
 
 
-def _decode_error(error_info: ErrorInfo, codec: Codec) -> BaseException:
-    """Convert ErrorInfo dict to exception."""
-    try:
-        if "state" not in error_info:
-            return Exception(f"[{error_info['code']}] {error_info['message']}")
-        return cast("BaseException", codec.decode_state(error_info["state"]))
-    except Exception:
-        return Exception(f"[{error_info['code']}] {error_info['message']}")
+def _decode_error(error_info: ErrorInfo) -> BaseException:
+    return Exception(f"[{error_info['code']}] {error_info['message']}")

tests/test_task.py

@@ -1,7 +1,11 @@
+from __future__ import annotations
+
 import asyncio
+import base64
+import pickle
 import random
 import uuid
-from typing import ClassVar
+from dataclasses import dataclass
 
 import pytest
 
@@ -104,31 +108,33 @@ async def activity(s: str) -> str:
     assert x == "hello"
 
 
-try:
-    import pydantic
+@dataclass
+class CustomPoint:
+    x: int
+    y: int
 
-    class Point(pydantic.BaseModel):
-        model_config: ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict(
-            extra="forbid"
-        )
 
-        x: int
-        y: int
+class PickleCodec:
+    def encode_json(self, result: object) -> str:
+        return base64.b64encode(pickle.dumps(result)).decode()
 
-    @pytest.mark.asyncio
-    async def test_pydantic():
-        @durable()
-        async def activity() -> Point:
-            ctx = get_context()
-            pt = await ctx.run(lambda: Point(x=1, y=2))
-            return Point(x=pt.x + 5, y=pt.y + 10)
+    def decode_json(self, encoded: object) -> object:
+        if not isinstance(encoded, str):
+            raise TypeError(f"Expected a string, got {type(encoded).__name__}")
+        return pickle.loads(base64.b64decode(encoded.encode()))
 
-        log = MemoryLogStorage()
-        async with task(activity, log) as t:
-            await t.start()
-            a = await t.wait()
-            assert type(a) is Point
-            assert a.x == 6 and a.y == 12
 
-except ImportError:
-    pass
+@pytest.mark.asyncio
+async def test_serialize():
+    @durable(codec=PickleCodec())
+    async def activity() -> CustomPoint:
+        ctx = get_context()
+        pt = await ctx.run(lambda: CustomPoint(x=1, y=2))
+        return CustomPoint(x=pt.x + 5, y=pt.y + 10)
+
+    log = MemoryLogStorage()
+    async with task(activity, log) as t:
+        await t.start()
+        a = await t.wait()
+        assert type(a) is CustomPoint
+        assert a.x == 6 and a.y == 12