Cleanup

Author: dandavison

temporalio/converter.py

@@ -136,6 +136,7 @@ class ActivitySerializationContext(BaseWorkflowSerializationContext):
     is_local: bool
 
 
+# TODO: duck typing or nominal typing?
 class WithSerializationContext(ABC):
     """Interface for objects that can use serialization context.
 
@@ -341,7 +342,6 @@ def __init__(self, *converters: EncodingPayloadConverter) -> None:
         Args:
             converters: Payload converters to delegate to, in order.
         """
-        # Insertion order preserved here since Python 3.7
         self.converters = {c.encoding.encode(): c for c in converters}
 
     def to_payloads(
@@ -407,7 +407,7 @@ def from_payloads(
         return values
 
     def with_context(self, context: SerializationContext) -> CompositePayloadConverter:
-        """Return a new instance with the given context."""
+        """Return a new instance with context set on the component converters"""
         return CompositePayloadConverter(
             *(
                 c.with_context(context)

temporalio/worker/_workflow_instance.py

@@ -2087,7 +2087,12 @@ def _converters_with_context(
         temporalio.converter.PayloadConverter,
         temporalio.converter.FailureConverter,
     ]:
-        """Construct workflow payload and failure converters with the given context."""
+        """Construct workflow payload and failure converters with the given context.
+
+        This plays a similar role to DataConverter._with_context, but operates on PayloadConverter
+        and FailureConverter only (since payload encoding/decoding is done by the worker, outside
+        the workflowsandbox).
+        """
         payload_converter = self._context_free_payload_converter
         failure_converter = self._context_free_failure_converter
         if isinstance(payload_converter, temporalio.converter.WithSerializationContext):

tests/test_serialization_context.py

@@ -21,9 +21,9 @@
 from pydantic import BaseModel
 from typing_extensions import Never
 
+import temporalio.api.common.v1
+import temporalio.api.failure.v1
 from temporalio import activity, workflow
-from temporalio.api.common.v1 import Payload
-from temporalio.api.failure.v1 import Failure
 from temporalio.client import Client, WorkflowFailureError, WorkflowUpdateFailedError
 from temporalio.common import RetryPolicy
 from temporalio.contrib.pydantic import PydanticJSONPlainPayloadConverter
@@ -82,7 +82,7 @@ def with_context(
         converter.context = context
         return converter
 
-    def to_payload(self, value: Any) -> Optional[Payload]:
+    def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:
         if not isinstance(value, TraceData):
             return None
         if isinstance(self.context, WorkflowSerializationContext):
@@ -106,7 +106,11 @@ def to_payload(self, value: Any) -> Optional[Payload]:
         payload.metadata["encoding"] = self.encoding.encode()
         return payload
 
-    def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> Any:
+    def from_payload(
+        self,
+        payload: temporalio.api.common.v1.Payload,
+        type_hint: Optional[Type] = None,
+    ) -> Any:
         value = JSONPlainPayloadConverter().from_payload(payload, TraceData)
         assert isinstance(value, TraceData)
         if isinstance(self.context, WorkflowSerializationContext):
@@ -1002,7 +1006,7 @@ def __init__(self):
 
     def with_context(
         self, context: Optional[SerializationContext]
-    ) -> "FailureConverterWithContext":
+    ) -> FailureConverterWithContext:
         converter = FailureConverterWithContext()
         converter.context = context
         return converter
@@ -1011,7 +1015,7 @@ def to_failure(
         self,
         exception: BaseException,
         payload_converter: PayloadConverter,
-        failure: Failure,
+        failure: temporalio.api.failure.v1.Failure,
     ) -> None:
         assert isinstance(
             self.context, (WorkflowSerializationContext, ActivitySerializationContext)
@@ -1025,7 +1029,9 @@ def to_failure(
         super().to_failure(exception, payload_converter, failure)
 
     def from_failure(
-        self, failure: Failure, payload_converter: PayloadConverter
+        self,
+        failure: temporalio.api.failure.v1.Failure,
+        payload_converter: PayloadConverter,
     ) -> BaseException:
         assert isinstance(
             self.context, (WorkflowSerializationContext, ActivitySerializationContext)
@@ -1132,12 +1138,14 @@ def __init__(self):
 
     def with_context(
         self, context: Optional[SerializationContext]
-    ) -> "PayloadCodecWithContext":
+    ) -> PayloadCodecWithContext:
         codec = PayloadCodecWithContext()
         codec.context = context
         return codec
 
-    async def encode(self, payloads: Sequence[Payload]) -> List[Payload]:
+    async def encode(
+        self, payloads: Sequence[temporalio.api.common.v1.Payload]
+    ) -> List[temporalio.api.common.v1.Payload]:
         assert self.context
         if isinstance(self.context, ActivitySerializationContext):
             test_traces[self.context.workflow_id].append(
@@ -1156,7 +1164,9 @@ async def encode(self, payloads: Sequence[Payload]) -> List[Payload]:
             )
         return list(payloads)
 
-    async def decode(self, payloads: Sequence[Payload]) -> List[Payload]:
+    async def decode(
+        self, payloads: Sequence[temporalio.api.common.v1.Payload]
+    ) -> List[temporalio.api.common.v1.Payload]:
         assert self.context
         if isinstance(self.context, ActivitySerializationContext):
             test_traces[self.context.workflow_id].append(
@@ -1439,20 +1449,24 @@ def with_context(
         codec.context = context
         return codec
 
-    async def encode(self, payloads: Sequence[Payload]) -> List[Payload]:
+    async def encode(
+        self, payloads: Sequence[temporalio.api.common.v1.Payload]
+    ) -> List[temporalio.api.common.v1.Payload]:
         [payload] = payloads
         return [
-            Payload(
+            temporalio.api.common.v1.Payload(
                 metadata=payload.metadata,
                 data=json.dumps(self._get_encryption_key()).encode(),
             )
         ]
 
-    async def decode(self, payloads: Sequence[Payload]) -> List[Payload]:
+    async def decode(
+        self, payloads: Sequence[temporalio.api.common.v1.Payload]
+    ) -> List[temporalio.api.common.v1.Payload]:
         [payload] = payloads
         assert json.loads(payload.data.decode()) == self._get_encryption_key()
         metadata = dict(payload.metadata)
-        return [Payload(metadata=metadata, data=b'"inbound"')]
+        return [temporalio.api.common.v1.Payload(metadata=metadata, data=b'"inbound"')]
 
     def _get_encryption_key(self) -> str:
         context = (
@@ -1594,8 +1608,8 @@ def with_context(
         return codec
 
     async def _assert_context_iff_not_nexus(
-        self, payloads: Sequence[Payload]
-    ) -> List[Payload]:
+        self, payloads: Sequence[temporalio.api.common.v1.Payload]
+    ) -> List[temporalio.api.common.v1.Payload]:
         [payload] = payloads
         assert bool(self.context) == (payload.data.decode() != '"nexus-data"')
         return list(payloads)
@@ -1670,12 +1684,12 @@ def __init__(self):
 
     def with_context(
         self, context: Optional[SerializationContext]
-    ) -> "PydanticJSONConverterWithContext":
+    ) -> PydanticJSONConverterWithContext:
         converter = PydanticJSONConverterWithContext()
         converter.context = context
         return converter
 
-    def to_payload(self, value: Any) -> Optional[Payload]:
+    def to_payload(self, value: Any) -> Optional[temporalio.api.common.v1.Payload]:
         if isinstance(value, PydanticData) and self.context:
             if isinstance(self.context, WorkflowSerializationContext):
                 value.trace.append(f"wf_{self.context.workflow_id}")