Make DataConverter implement the interface

Author: dandavison

temporalio/client.py

@@ -1609,7 +1609,7 @@ def __init__(
 
     @functools.cached_property
     def _data_converter(self) -> temporalio.converter.DataConverter:
-        return self._client.data_converter._with_context(
+        return self._client.data_converter.with_context(
             temporalio.converter.WorkflowSerializationContext(
                 namespace=self._client.namespace, workflow_id=self._id
             )
@@ -2865,7 +2865,7 @@ def with_context(self, context: SerializationContext) -> AsyncActivityHandle:
         return AsyncActivityHandle(
             self._client,
             self._id_or_token,
-            self._client.data_converter._with_context(context),
+            self._client.data_converter.with_context(context),
         )
 
 
@@ -3234,7 +3234,7 @@ async def fetch_next_page(self, *, page_size: Optional[int] = None) -> None:
         def get_data_converter(workflow_id: str) -> temporalio.converter.DataConverter:
             if workflow_id not in data_converter_cache:
                 data_converter_cache[workflow_id] = (
-                    self._client.data_converter._with_context(
+                    self._client.data_converter.with_context(
                         WorkflowSerializationContext(
                             namespace=self._client.namespace,
                             workflow_id=workflow_id,
@@ -4204,7 +4204,7 @@ async def _to_proto(
         priority: Optional[temporalio.api.common.v1.Priority] = None
         if self.priority:
             priority = self.priority._to_proto()
-        data_converter = client.data_converter._with_context(
+        data_converter = client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=client.namespace,
                 workflow_id=self.id,
@@ -5063,7 +5063,7 @@ def __init__(
 
     @functools.cached_property
     def _data_converter(self) -> temporalio.converter.DataConverter:
-        return self._client.data_converter._with_context(
+        return self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=self.workflow_id,
@@ -5977,7 +5977,7 @@ async def _build_signal_with_start_workflow_execution_request(
         self, input: StartWorkflowInput
     ) -> temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest:
         assert input.start_signal
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=input.id,
@@ -6008,7 +6008,7 @@ async def _populate_start_workflow_execution_request(
         ],
         input: Union[StartWorkflowInput, UpdateWithStartStartWorkflowInput],
     ) -> None:
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=input.id,
@@ -6095,7 +6095,7 @@ async def describe_workflow(
                 metadata=input.rpc_metadata,
                 timeout=input.rpc_timeout,
             ),
-            self._client.data_converter._with_context(
+            self._client.data_converter.with_context(
                 WorkflowSerializationContext(
                     namespace=self._client.namespace,
                     workflow_id=input.id,
@@ -6129,7 +6129,7 @@ async def count_workflows(
         )
 
     async def query_workflow(self, input: QueryWorkflowInput) -> Any:
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=input.id,
@@ -6182,7 +6182,7 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any:
         return results[0]
 
     async def signal_workflow(self, input: SignalWorkflowInput) -> None:
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=input.id,
@@ -6207,7 +6207,7 @@ async def signal_workflow(self, input: SignalWorkflowInput) -> None:
         )
 
     async def terminate_workflow(self, input: TerminateWorkflowInput) -> None:
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=input.id,
@@ -6281,7 +6281,7 @@ async def _build_update_workflow_execution_request(
         input: Union[StartWorkflowUpdateInput, UpdateWithStartUpdateWorkflowInput],
         workflow_id: str,
     ) -> temporalio.api.workflowservice.v1.UpdateWorkflowExecutionRequest:
-        data_converter = self._client.data_converter._with_context(
+        data_converter = self._client.data_converter.with_context(
             WorkflowSerializationContext(
                 namespace=self._client.namespace,
                 workflow_id=workflow_id,

temporalio/converter.py

@@ -1233,7 +1233,7 @@ def __init__(self) -> None:
 
 
 @dataclass(frozen=True)
-class DataConverter:
+class DataConverter(WithSerializationContext):
     """Data converter for converting and encoding payloads to/from Python values.
 
     This combines :py:class:`PayloadConverter` which converts values with
@@ -1337,7 +1337,8 @@ async def decode_failure(
             await self.payload_codec.decode_failure(failure)
         return self.failure_converter.from_failure(failure, self.payload_converter)
 
-    def _with_context(self, context: SerializationContext) -> Self:
+    def with_context(self, context: SerializationContext) -> Self:
+        """Return an instance with context set on the component converters."""
         payload_converter = self.payload_converter
         payload_codec = self.payload_codec
         failure_converter = self.failure_converter

temporalio/worker/_activity.py

@@ -262,7 +262,7 @@ async def _heartbeat_async(
                 activity_task_queue=self._task_queue,
                 is_local=activity.info.is_local,
             )
-            data_converter = data_converter._with_context(context)
+            data_converter = data_converter.with_context(context)
 
         # Perform the heartbeat
         try:
@@ -314,7 +314,7 @@ async def _handle_start_activity_task(
             activity_task_queue=self._task_queue,
             is_local=start.is_local,
         )
-        data_converter = self._data_converter._with_context(context)
+        data_converter = self._data_converter.with_context(context)
         try:
             result = await self._execute_activity(
                 start, running_activity, task_token, data_converter

temporalio/worker/_workflow.py

@@ -275,7 +275,7 @@ async def _handle_activation(
                         "Cache already exists for activation with initialize job"
                     )
 
-            data_converter = self._data_converter._with_context(
+            data_converter = self._data_converter.with_context(
                 temporalio.converter.WorkflowSerializationContext(
                     namespace=self._namespace,
                     workflow_id=workflow_id,