Memoize patched calls and support UUID conversion (#192)

Author: cretz

README.md

@@ -189,7 +189,11 @@ The default data converter supports converting multiple types including:
   * Iterables including ones JSON dump may not support by default, e.g. `set`
   * Any class with a `dict()` method and a static `parse_obj()` method, e.g.
     [Pydantic models](https://pydantic-docs.helpmanual.io/usage/models)
+    * Note, this doesn't mean every Pydantic field can be converted, only fields which the data converter supports
   * [IntEnum, StrEnum](https://docs.python.org/3/library/enum.html) based enumerates
+  * [UUID](https://docs.python.org/3/library/uuid.html)
+
+This notably doesn't include any `date`, `time`, or `datetime` objects as they may not work across SDKs.
 
 For converting from JSON, the workflow/activity type hint is taken into account to convert to the proper type. Care has
 been taken to support all common typings including `Optional`, `Union`, all forms of iterables and mappings, `NewType`,

temporalio/converter.py

@@ -9,6 +9,7 @@
 import json
 import sys
 import traceback
+import uuid
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from datetime import datetime
@@ -426,6 +427,9 @@ def default(self, o: Any) -> Any:
         # Support for non-list iterables like set
         if not isinstance(o, list) and isinstance(o, collections.abc.Iterable):
             return list(o)
+        # Support for UUID
+        if isinstance(o, uuid.UUID):
+            return str(o)
         return super().default(o)
 
 
@@ -1273,6 +1277,10 @@ def value_to_type(hint: Type, value: Any) -> Any:
                 )
             return hint(value)
 
+    # UUID
+    if inspect.isclass(hint) and issubclass(hint, uuid.UUID):
+        return hint(value)
+
     # Iterable. We intentionally put this last as it catches several others.
     if inspect.isclass(origin) and issubclass(origin, collections.abc.Iterable):
         if not isinstance(value, collections.abc.Iterable):

temporalio/worker/_workflow_instance.py

@@ -191,9 +191,9 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         self._is_replaying: bool = False
         self._random = random.Random(det.randomness_seed)
 
-        # Patches we have been notified of and patches that have been sent
+        # Patches we have been notified of and memoized patch responses
         self._patches_notified: Set[str] = set()
-        self._patches_sent: Set[str] = set()
+        self._patches_memoized: Dict[str, bool] = {}
 
         # Tasks stored by asyncio are weak references and therefore can get GC'd
         # which can cause warnings like "Task was destroyed but it is pending!".
@@ -739,13 +739,18 @@ def workflow_memo_value(
         )[0]
 
     def workflow_patch(self, id: str, *, deprecated: bool) -> bool:
+        # We use a previous memoized result of this if present. If this is being
+        # deprecated, we can still use memoized result and skip the command.
+        use_patch = self._patches_memoized.get(id)
+        if use_patch is not None:
+            return use_patch
+
         use_patch = not self._is_replaying or id in self._patches_notified
-        # Only add patch command if never sent before for this ID
-        if use_patch and not id in self._patches_sent:
+        self._patches_memoized[id] = use_patch
+        if use_patch:
             command = self._add_command()
             command.set_patch_marker.patch_id = id
             command.set_patch_marker.deprecated = deprecated
-            self._patches_sent.add(id)
         return use_patch
 
     def workflow_random(self) -> random.Random:

tests/test_converter.py

@@ -25,6 +25,7 @@
     Type,
     Union,
 )
+from uuid import UUID, uuid4
 
 import pydantic
 import pytest
@@ -224,6 +225,7 @@ class NestedDataClass:
     foo: str
     bar: List[NestedDataClass] = dataclasses.field(default_factory=list)
     baz: Optional[NestedDataClass] = None
+    qux: Optional[UUID] = None
 
 
 class MyTypedDict(TypedDict):
@@ -239,6 +241,7 @@ class MyTypedDictNotTotal(TypedDict, total=False):
 class MyPydanticClass(pydantic.BaseModel):
     foo: str
     bar: List[MyPydanticClass]
+    baz: Optional[UUID] = None
 
 
 def test_json_type_hints():
@@ -287,6 +290,7 @@ def fail(hint: Type, value: Any) -> None:
     ok(NestedDataClass, NestedDataClass("foo"))
     ok(NestedDataClass, NestedDataClass("foo", baz=NestedDataClass("bar")))
     ok(NestedDataClass, NestedDataClass("foo", bar=[NestedDataClass("bar")]))
+    ok(NestedDataClass, NestedDataClass("foo", qux=uuid4()))
     # Missing required dataclass fields causes failure
     ok(NestedDataClass, {"foo": "bar"}, NestedDataClass("bar"))
     fail(NestedDataClass, {})
@@ -346,6 +350,8 @@ def fail(hint: Type, value: Any) -> None:
     ok(SerializableEnum, SerializableEnum.FOO)
     ok(List[SerializableEnum], [SerializableEnum.FOO, SerializableEnum.FOO])
 
+    # UUID
+
     # StrEnum is available in 3.11+
     if sys.version_info >= (3, 11):
         # StrEnum
@@ -364,7 +370,9 @@ def fail(hint: Type, value: Any) -> None:
     # Pydantic
     ok(
         MyPydanticClass,
-        MyPydanticClass(foo="foo", bar=[MyPydanticClass(foo="baz", bar=[])]),
+        MyPydanticClass(
+            foo="foo", bar=[MyPydanticClass(foo="baz", bar=[])], baz=uuid4()
+        ),
     )
     ok(List[MyPydanticClass], [MyPydanticClass(foo="foo", bar=[])])
     fail(List[MyPydanticClass], [MyPydanticClass(foo="foo", bar=[]), 5])

tests/worker/test_workflow.py

@@ -67,6 +67,7 @@
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import (
     UnsandboxedWorkflowRunner,
+    Worker,
     WorkflowInstance,
     WorkflowInstanceDetails,
     WorkflowRunner,
@@ -2069,6 +2070,87 @@ async def query_result(handle: WorkflowHandle) -> str:
         # await query_result(patch_handle)
 
 
+@workflow.defn(name="patch-memoized")
+class PatchMemoizedWorkflowUnpatched:
+    def __init__(self, *, should_patch: bool = False) -> None:
+        self.should_patch = should_patch
+        self._waiting_signal = True
+
+    @workflow.run
+    async def run(self) -> List[str]:
+        results: List[str] = []
+        if self.should_patch and workflow.patched("some-patch"):
+            results.append("pre-patch")
+        self._waiting_signal = True
+        await workflow.wait_condition(lambda: not self._waiting_signal)
+        results.append("some-value")
+        if self.should_patch and workflow.patched("some-patch"):
+            results.append("post-patch")
+        return results
+
+    @workflow.signal
+    def signal(self) -> None:
+        self._waiting_signal = False
+
+    @workflow.query
+    def waiting_signal(self) -> bool:
+        return self._waiting_signal
+
+
+@workflow.defn(name="patch-memoized")
+class PatchMemoizedWorkflowPatched(PatchMemoizedWorkflowUnpatched):
+    def __init__(self) -> None:
+        super().__init__(should_patch=True)
+
+    @workflow.run
+    async def run(self) -> List[str]:
+        return await super().run()
+
+
+async def test_workflow_patch_memoized(client: Client):
+    # Start a worker with the workflow unpatched and wait until halfway through
+    task_queue = f"tq-{uuid.uuid4()}"
+    async with Worker(
+        client, task_queue=task_queue, workflows=[PatchMemoizedWorkflowUnpatched]
+    ):
+        pre_patch_handle = await client.start_workflow(
+            PatchMemoizedWorkflowUnpatched.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+
+        # Need to wait until it has gotten halfway through
+        async def waiting_signal() -> bool:
+            return await pre_patch_handle.query(
+                PatchMemoizedWorkflowUnpatched.waiting_signal
+            )
+
+        await assert_eq_eventually(True, waiting_signal)
+
+    # Now start the worker again, but this time with a patched workflow
+    async with Worker(
+        client, task_queue=task_queue, workflows=[PatchMemoizedWorkflowPatched]
+    ):
+        # Start a new workflow post patch
+        post_patch_handle = await client.start_workflow(
+            PatchMemoizedWorkflowPatched.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+
+        # Send signal to both and check results
+        await pre_patch_handle.signal(PatchMemoizedWorkflowPatched.signal)
+        await post_patch_handle.signal(PatchMemoizedWorkflowPatched.signal)
+
+        # Confirm expected values
+        assert ["some-value"] == await pre_patch_handle.result()
+        assert [
+            "pre-patch",
+            "some-value",
+            "post-patch",
+        ] == await post_patch_handle.result()
+
+
 @workflow.defn
 class UUIDWorkflow:
     def __init__(self) -> None: