Construct context-aware data converter lazily on WorkflowExecution

Author: dandavison

temporalio/client.py

@@ -2883,10 +2883,6 @@ class WorkflowExecution:
     close_time: Optional[datetime]
     """When the workflow was closed if closed."""
 
-    # WorkflowExecution.data_converter
-    data_converter: temporalio.converter.DataConverter
-    """Data converter from when this description was created."""
-
     execution_time: Optional[datetime]
     """When this workflow run started or should start."""
 
@@ -2896,6 +2892,9 @@ class WorkflowExecution:
     id: str
     """ID for the workflow."""
 
+    namespace: str
+    """Namespace for the workflow."""
+
     parent_id: Optional[str]
     """ID for the parent workflow if this was started as a child."""
 
@@ -2936,10 +2935,22 @@ class WorkflowExecution:
     workflow_type: str
     """Type name for the workflow."""
 
+    _context_free_data_converter: temporalio.converter.DataConverter
+
+    @property
+    def data_converter(self) -> temporalio.converter.DataConverter:
+        return self._context_free_data_converter.with_context(
+            WorkflowSerializationContext(
+                namespace=self.namespace,
+                workflow_id=self.id,
+            )
+        )
+
     @classmethod
     def _from_raw_info(
         cls,
         info: temporalio.api.workflow.v1.WorkflowExecutionInfo,
+        namespace: str,
         converter: temporalio.converter.DataConverter,
         **additional_fields: Any,
     ) -> WorkflowExecution:
@@ -2949,14 +2960,14 @@ def _from_raw_info(
                 if info.HasField("close_time")
                 else None
             ),
-            data_converter=converter,
             execution_time=(
                 info.execution_time.ToDatetime().replace(tzinfo=timezone.utc)
                 if info.HasField("execution_time")
                 else None
             ),
             history_length=info.history_length,
             id=info.execution.workflow_id,
+            namespace=namespace,
             parent_id=(
                 info.parent_execution.workflow_id
                 if info.HasField("parent_execution")
@@ -2987,6 +2998,7 @@ def _from_raw_info(
                 info.search_attributes
             ),
             workflow_type=info.type.name,
+            _context_free_data_converter=converter,
             **additional_fields,
         )
 
@@ -3247,23 +3259,9 @@ async def fetch_next_page(self, *, page_size: Optional[int] = None) -> None:
             timeout=self._input.rpc_timeout,
         )
 
-        data_converter_cache = {}
-
-        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(
-                        WorkflowSerializationContext(
-                            namespace=self._client.namespace,
-                            workflow_id=workflow_id,
-                        )
-                    )
-                )
-            return data_converter_cache[workflow_id]
-
         self._current_page = [
             WorkflowExecution._from_raw_info(
-                v, get_data_converter(v.execution.workflow_id)
+                v, self._client.namespace, self._client.data_converter
             )
             for v in resp.executions
         ]