From b61dad7e03230f12444b125daddf5e6d4caf1747 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 16 Jul 2026 11:53:50 -0700 Subject: [PATCH 1/3] Wrap payload converters for temporal intermediate models --- temporalio/activity.py | 12 +++- temporalio/converter/__init__.py | 2 + temporalio/converter/_data_converter.py | 7 +- temporalio/converter/_payload_converter.py | 77 ++++++++++++++++++++++ temporalio/worker/_workflow_instance.py | 6 +- tests/test_converter.py | 54 +++++++++++++++ 6 files changed, 154 insertions(+), 4 deletions(-) diff --git a/temporalio/activity.py b/temporalio/activity.py index 4e632701e..1c80438e9 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -238,9 +238,17 @@ def payload_converter(self) -> temporalio.converter.PayloadConverter: self.payload_converter_class_or_instance, temporalio.converter.PayloadConverter, ): - self._payload_converter = self.payload_converter_class_or_instance + self._payload_converter = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + self.payload_converter_class_or_instance + ) + ) else: - self._payload_converter = self.payload_converter_class_or_instance() + self._payload_converter = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + self.payload_converter_class_or_instance() + ) + ) return self._payload_converter @property diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 3821cbd68..7cdec2a76 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -33,6 +33,7 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadConverter, + TemporalIntermediatePayloadConverter, value_to_type, ) from temporalio.converter._payload_limits import ( @@ -83,6 +84,7 @@ "PayloadLimitsConfig", "PayloadSizeWarning", "SerializationContext", + "TemporalIntermediatePayloadConverter", "WithSerializationContext", "WorkflowSerializationContext", "decode_search_attributes", diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 13b48e695..b0c87b926 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -29,6 +29,7 @@ ) from temporalio.converter._payload_converter import ( PayloadConverter, + TemporalIntermediatePayloadConverter, ) from temporalio.converter._payload_limits import ( PayloadLimitsConfig, @@ -103,7 +104,11 @@ class DataConverter(WithSerializationContext): """Server-reported limits for payloads.""" def __post_init__(self) -> None: # noqa: D105 - object.__setattr__(self, "payload_converter", self.payload_converter_class()) + object.__setattr__( + self, + "payload_converter", + TemporalIntermediatePayloadConverter.wrap(self.payload_converter_class()), + ) object.__setattr__(self, "failure_converter", self.failure_converter_class()) async def encode( diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 8ee85ef72..261ad1033 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -514,6 +514,83 @@ def from_payload( raise RuntimeError("Failed parsing") from err +class TemporalIntermediatePayloadConverter(PayloadConverter, WithSerializationContext): + """Payload converter wrapper for generated Temporal intermediate hooks. + + Values with a ``_temporal_to_intermediate`` method are first converted to + their intermediate value, then encoded by the wrapped payload converter. When + decoding to a type with ``_temporal_from_intermediate``, the wrapped + converter first decodes the payload to the intermediate value and this + wrapper constructs the requested user-facing type from it. + """ + + _inner_payload_converter: PayloadConverter + + def __init__(self, inner_payload_converter: PayloadConverter) -> None: + """Create a Temporal intermediate payload converter.""" + self._inner_payload_converter = inner_payload_converter + + @staticmethod + def wrap(payload_converter: PayloadConverter) -> PayloadConverter: + """Wrap a payload converter unless it is already wrapped.""" + if isinstance(payload_converter, TemporalIntermediatePayloadConverter): + return payload_converter + return TemporalIntermediatePayloadConverter(payload_converter) + + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + """See base class.""" + intermediate_values: list[Any] = [] + for value in values: + to_intermediate = getattr(value, "_temporal_to_intermediate", None) + if to_intermediate is not None: + value = to_intermediate(payload_converter=self._inner_payload_converter) + intermediate_values.append(value) + return self._inner_payload_converter.to_payloads(intermediate_values) + + def from_payloads( + self, + payloads: Sequence[temporalio.api.common.v1.Payload], + type_hints: list[type] | None = None, + ) -> list[Any]: + """See base class.""" + if type_hints is None: + return self._inner_payload_converter.from_payloads(payloads, None) + normalized_type_hints: list[type | None] = list(type_hints) + if len(normalized_type_hints) < len(payloads): + normalized_type_hints.extend([None] * (len(payloads) - len(type_hints))) + inner_type_hints = [ + None + if getattr(type_hint, "_temporal_from_intermediate", None) is not None + else type_hint + for type_hint in normalized_type_hints + ] + values = self._inner_payload_converter.from_payloads( + payloads, typing.cast("list[type]", inner_type_hints) + ) + return [ + from_intermediate(value, payload_converter=self._inner_payload_converter) + if ( + from_intermediate := getattr( + type_hint, "_temporal_from_intermediate", None + ) + ) + is not None + else value + for value, type_hint in zip(values, normalized_type_hints) + ] + + def with_context(self, context: SerializationContext) -> Self: + """Return a new instance with context set on the inner converter.""" + if not isinstance(self._inner_payload_converter, WithSerializationContext): + return self + inner_payload_converter = self._inner_payload_converter.with_context(context) + if inner_payload_converter is self._inner_payload_converter: + return self + return type(self)(inner_payload_converter) + + class AdvancedJSONEncoder(json.JSONEncoder): """Advanced JSON encoder. diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 74edc66b7..4d329059b 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -246,7 +246,11 @@ def __init__(self, det: WorkflowInstanceDetails) -> None: self._defn = det.defn self._workflow_input: ExecuteWorkflowInput | None = None self._info = det.info - self._context_free_payload_converter = det.payload_converter_class() + self._context_free_payload_converter = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + det.payload_converter_class() + ) + ) self._context_free_failure_converter = det.failure_converter_class() workflow_context = temporalio.converter.WorkflowSerializationContext( namespace=det.info.namespace, diff --git a/tests/test_converter.py b/tests/test_converter.py index 10365f9c1..2281513f2 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -44,6 +44,8 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, + PayloadConverter, + TemporalIntermediatePayloadConverter, decode_search_attributes, encode_search_attribute_values, value_to_type, @@ -254,6 +256,58 @@ def test_binary_proto(): assert decoded == proto +@dataclass +class TemporalIntermediateModel: + value: str + + @classmethod + def _temporal_from_intermediate( + cls, + intermediate: temporalio.api.common.v1.WorkflowExecution, + *, + payload_converter: PayloadConverter | None = None, + ) -> TemporalIntermediateModel: + assert payload_converter is not None + return cls(value=intermediate.workflow_id) + + def _temporal_to_intermediate( + self, *, payload_converter: PayloadConverter | None = None + ) -> temporalio.api.common.v1.WorkflowExecution: + assert payload_converter is not None + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=self.value, + run_id="run-id", + ) + + +class CustomDefaultPayloadConverter(DefaultPayloadConverter): + pass + + +def test_temporal_intermediate_payload_converter_wraps_user_converter(): + data_converter = DataConverter( + payload_converter_class=CustomDefaultPayloadConverter + ) + converter = data_converter.payload_converter + assert isinstance(converter, TemporalIntermediatePayloadConverter) + value = TemporalIntermediateModel("workflow-id") + + payload = converter.to_payload(value) + + assert payload.metadata["encoding"] == b"json/protobuf" + assert ( + payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" + ) + assert all("temporal-wire" not in key for key in payload.metadata) + assert all(b"temporal-wire" not in value for value in payload.metadata.values()) + assert converter.from_payload(payload, TemporalIntermediateModel) == value + + plain_proto_payload = converter.to_payload( + temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2") + ) + assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" + + def test_encode_search_attribute_values(): with pytest.raises(TypeError, match="of type tuple not one of"): encode_search_attribute_values([("bad type",)]) # type: ignore[arg-type] From 585bc27d62e9dc5f299685289ef239c32bcb215a Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 16 Jul 2026 13:04:40 -0700 Subject: [PATCH 2/3] Support intermediate hooks in system Nexus conversion --- temporalio/converter/_payload_converter.py | 4 +- temporalio/nexus/system/__init__.py | 74 ++++++++++++++++++++-- temporalio/worker/_workflow_instance.py | 4 +- tests/nexus/test_temporal_system_nexus.py | 12 +++- tests/test_converter.py | 8 +-- tests/worker/test_visitor.py | 4 +- 6 files changed, 87 insertions(+), 19 deletions(-) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 261ad1033..8cdcc7abe 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -545,7 +545,7 @@ def to_payloads( for value in values: to_intermediate = getattr(value, "_temporal_to_intermediate", None) if to_intermediate is not None: - value = to_intermediate(payload_converter=self._inner_payload_converter) + value = to_intermediate() intermediate_values.append(value) return self._inner_payload_converter.to_payloads(intermediate_values) @@ -570,7 +570,7 @@ def from_payloads( payloads, typing.cast("list[type]", inner_type_hints) ) return [ - from_intermediate(value, payload_converter=self._inner_payload_converter) + from_intermediate(value) if ( from_intermediate := getattr( type_hint, "_temporal_from_intermediate", None diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 21c5a1408..e76756cd7 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -2,22 +2,82 @@ from __future__ import annotations +import contextlib +import contextvars +from collections.abc import Iterator, Sequence +from typing import Any + import temporalio.api.common.v1 import temporalio.converter from temporalio.bridge._visitor_functions import VisitorFunctions from temporalio.converter import BinaryProtoPayloadConverter, CompositePayloadConverter TEMPORAL_SYSTEM_ENDPOINT = "__temporal_system" +_user_payload_converter: contextvars.ContextVar[ + temporalio.converter.PayloadConverter | None +] = contextvars.ContextVar("temporal-system-nexus-user-payload-converter", default=None) -class SystemNexusPayloadConverter(CompositePayloadConverter): - """Payload converter for system Nexus outer envelopes.""" +@contextlib.contextmanager +def user_payload_converter_context( + payload_converter: temporalio.converter.PayloadConverter, +) -> Iterator[None]: + """Set the user payload converter for system Nexus model conversion.""" + token = _user_payload_converter.set(payload_converter) + try: + yield + finally: + _user_payload_converter.reset(token) + + +def current_user_payload_converter() -> temporalio.converter.PayloadConverter: + """Return the active user payload converter for system Nexus model conversion.""" + payload_converter = _user_payload_converter.get() + if payload_converter is None: + raise RuntimeError("System Nexus user payload converter context is not active") + return payload_converter + + +class _SystemNexusOuterPayloadConverter(CompositePayloadConverter): + """Payload converter for system Nexus outer proto envelopes.""" def __init__(self) -> None: """Create a payload converter for system Nexus outer envelopes.""" super().__init__(BinaryProtoPayloadConverter()) +class SystemNexusPayloadConverter(temporalio.converter.PayloadConverter): + """Payload converter for system Nexus outer envelopes.""" + + _user_payload_converter: temporalio.converter.PayloadConverter + _outer_payload_converter: temporalio.converter.PayloadConverter + + def __init__(self, user_payload_converter: temporalio.converter.PayloadConverter) -> None: + """Create a payload converter for system Nexus outer envelopes.""" + self._user_payload_converter = user_payload_converter + self._outer_payload_converter = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + _SystemNexusOuterPayloadConverter() + ) + ) + + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + """See base class.""" + with user_payload_converter_context(self._user_payload_converter): + return self._outer_payload_converter.to_payloads(values) + + def from_payloads( + self, + payloads: Sequence[temporalio.api.common.v1.Payload], + type_hints: list[type] | None = None, + ) -> list[Any]: + """See base class.""" + with user_payload_converter_context(self._user_payload_converter): + return self._outer_payload_converter.from_payloads(payloads, type_hints) + + def is_system_endpoint(endpoint: str) -> bool: """Return whether a Nexus endpoint is the Temporal system endpoint.""" return endpoint == TEMPORAL_SYSTEM_ENDPOINT @@ -33,7 +93,7 @@ async def maybe_visit_payload( if not is_system_endpoint(endpoint): return None - payload_converter = get_payload_converter() + payload_converter = _SystemNexusOuterPayloadConverter() value = payload_converter.from_payload(payload) from ._payload_visitor import PayloadVisitor @@ -43,15 +103,19 @@ async def maybe_visit_payload( return payload_converter.to_payload(value) -def get_payload_converter() -> temporalio.converter.PayloadConverter: +def get_payload_converter( + user_payload_converter: temporalio.converter.PayloadConverter, +) -> temporalio.converter.PayloadConverter: """Return the fixed payload converter for system Nexus outer envelopes.""" - return SystemNexusPayloadConverter() + return SystemNexusPayloadConverter(user_payload_converter) __all__ = [ "TEMPORAL_SYSTEM_ENDPOINT", + "current_user_payload_converter", "get_payload_converter", "is_system_endpoint", "maybe_visit_payload", "SystemNexusPayloadConverter", + "user_payload_converter_context", ] diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 4d329059b..dd7f4571a 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2093,7 +2093,9 @@ async def operation_handle_fn() -> OutputT: t.uncancel() # type: ignore[union-attr] payload_converter = ( - temporalio.nexus.system.get_payload_converter() + temporalio.nexus.system.get_payload_converter( + self._workflow_context_payload_converter + ) if temporalio.nexus.system.is_system_endpoint(input.endpoint) else self._context_free_payload_converter ) diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index b689ee8d9..4629d230b 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -186,7 +186,9 @@ def _new_system_nexus_request_payload() -> temporalio.api.common.v1.Payload: assert nested_payload is not None request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest() request.input.payloads.add().CopyFrom(nested_payload) - payload = nexus_system.get_payload_converter().to_payload(request) + payload = nexus_system.get_payload_converter( + temporalio.converter.PayloadConverter.default + ).to_payload(request) assert payload is not None return payload @@ -201,7 +203,9 @@ async def test_schedule_system_nexus_endpoint_ignores_operation_registry() -> No await PayloadVisitor().visit(visitor, completion) schedule = completion.successful.commands[0].schedule_nexus_operation - decoded = nexus_system.get_payload_converter().from_payload(schedule.input) + decoded = nexus_system.get_payload_converter( + temporalio.converter.PayloadConverter.default + ).from_payload(schedule.input) assert isinstance( decoded, workflowservice_pb2.SignalWithStartWorkflowExecutionRequest ) @@ -339,7 +343,9 @@ def _field_is_repeated(field: FieldDescriptor) -> bool: ], ) def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: - payload_converter = nexus_system.get_payload_converter() + payload_converter = nexus_system.get_payload_converter( + temporalio.converter.PayloadConverter.default + ) proto_value = _build_proto_sample(message_type) payload = payload_converter.to_payload(proto_value) assert payload is not None diff --git a/tests/test_converter.py b/tests/test_converter.py index 2281513f2..0ae2a2464 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -264,16 +264,10 @@ class TemporalIntermediateModel: def _temporal_from_intermediate( cls, intermediate: temporalio.api.common.v1.WorkflowExecution, - *, - payload_converter: PayloadConverter | None = None, ) -> TemporalIntermediateModel: - assert payload_converter is not None return cls(value=intermediate.workflow_id) - def _temporal_to_intermediate( - self, *, payload_converter: PayloadConverter | None = None - ) -> temporalio.api.common.v1.WorkflowExecution: - assert payload_converter is not None + def _temporal_to_intermediate(self) -> temporalio.api.common.v1.WorkflowExecution: return temporalio.api.common.v1.WorkflowExecution( workflow_id=self.value, run_id="run-id", diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index bd4004625..aa9a931e1 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -357,7 +357,9 @@ async def _visit(self) -> None: finally: active_visits -= 1 - payload_converter = nexus_system.get_payload_converter() + payload_converter = nexus_system.get_payload_converter( + temporalio.converter.PayloadConverter.default + ) system_request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest( input=Payloads(payloads=[Payload(data=b"workflow-input")]), signal_input=Payloads(payloads=[Payload(data=b"signal-input")]), From 844deb00b8238383f99e30a97c613ba7c0c5baee Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 21 Jul 2026 08:30:59 -0700 Subject: [PATCH 3/3] Rename intermediate hooks to data model hooks --- CHANGELOG.md | 6 +++ temporalio/activity.py | 13 +++---- temporalio/converter/__init__.py | 2 - temporalio/converter/_data_converter.py | 12 +++--- temporalio/converter/_payload_converter.py | 38 +++++++++---------- temporalio/nexus/system/__init__.py | 11 +++--- temporalio/worker/_workflow.py | 2 +- temporalio/worker/_workflow_instance.py | 8 +--- temporalio/worker/workflow_sandbox/_runner.py | 2 +- tests/test_converter.py | 15 ++++---- 10 files changed, 52 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a1a92a95..013583ebd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,12 @@ to include examples, links to docs, or any other relevant information. ### Added +- Added SDK payload converter support for values and type hints that expose + `_temporal_to_data_model` and `_temporal_from_data_model` hooks. This lets + hook-aware types delegate their wire representation to the configured payload + converter, preserving SDK behavior such as serialization contexts for nested + payload fields. + ### Changed ### Deprecated diff --git a/temporalio/activity.py b/temporalio/activity.py index 1c80438e9..a51a8c007 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -29,6 +29,7 @@ import temporalio.bridge.proto.activity_task import temporalio.common import temporalio.converter +from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter from .types import CallableType @@ -238,16 +239,12 @@ def payload_converter(self) -> temporalio.converter.PayloadConverter: self.payload_converter_class_or_instance, temporalio.converter.PayloadConverter, ): - self._payload_converter = ( - temporalio.converter.TemporalIntermediatePayloadConverter.wrap( - self.payload_converter_class_or_instance - ) + self._payload_converter = _TemporalDataModelPayloadConverter.wrap( + self.payload_converter_class_or_instance ) else: - self._payload_converter = ( - temporalio.converter.TemporalIntermediatePayloadConverter.wrap( - self.payload_converter_class_or_instance() - ) + self._payload_converter = _TemporalDataModelPayloadConverter.wrap( + self.payload_converter_class_or_instance() ) return self._payload_converter diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 7cdec2a76..3821cbd68 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -33,7 +33,6 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadConverter, - TemporalIntermediatePayloadConverter, value_to_type, ) from temporalio.converter._payload_limits import ( @@ -84,7 +83,6 @@ "PayloadLimitsConfig", "PayloadSizeWarning", "SerializationContext", - "TemporalIntermediatePayloadConverter", "WithSerializationContext", "WorkflowSerializationContext", "decode_search_attributes", diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index b0c87b926..0832cd182 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -29,7 +29,7 @@ ) from temporalio.converter._payload_converter import ( PayloadConverter, - TemporalIntermediatePayloadConverter, + _TemporalDataModelPayloadConverter, ) from temporalio.converter._payload_limits import ( PayloadLimitsConfig, @@ -104,13 +104,13 @@ class DataConverter(WithSerializationContext): """Server-reported limits for payloads.""" def __post_init__(self) -> None: # noqa: D105 - object.__setattr__( - self, - "payload_converter", - TemporalIntermediatePayloadConverter.wrap(self.payload_converter_class()), - ) + object.__setattr__(self, "payload_converter", self._new_payload_converter()) object.__setattr__(self, "failure_converter", self.failure_converter_class()) + def _new_payload_converter(self) -> PayloadConverter: + """Create a payload converter instance with SDK data model hooks enabled.""" + return _TemporalDataModelPayloadConverter.wrap(self.payload_converter_class()) + async def encode( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 8cdcc7abe..7d29d65b8 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -514,40 +514,40 @@ def from_payload( raise RuntimeError("Failed parsing") from err -class TemporalIntermediatePayloadConverter(PayloadConverter, WithSerializationContext): - """Payload converter wrapper for generated Temporal intermediate hooks. +class _TemporalDataModelPayloadConverter(PayloadConverter, WithSerializationContext): + """Payload converter wrapper for generated Temporal data model hooks. - Values with a ``_temporal_to_intermediate`` method are first converted to - their intermediate value, then encoded by the wrapped payload converter. When - decoding to a type with ``_temporal_from_intermediate``, the wrapped - converter first decodes the payload to the intermediate value and this + Values with a ``_temporal_to_data_model`` method are first converted to + their data model value, then encoded by the wrapped payload converter. When + decoding to a type with ``_temporal_from_data_model``, the wrapped + converter first decodes the payload to the data model value and this wrapper constructs the requested user-facing type from it. """ _inner_payload_converter: PayloadConverter def __init__(self, inner_payload_converter: PayloadConverter) -> None: - """Create a Temporal intermediate payload converter.""" + """Create a Temporal data model payload converter.""" self._inner_payload_converter = inner_payload_converter @staticmethod def wrap(payload_converter: PayloadConverter) -> PayloadConverter: """Wrap a payload converter unless it is already wrapped.""" - if isinstance(payload_converter, TemporalIntermediatePayloadConverter): + if isinstance(payload_converter, _TemporalDataModelPayloadConverter): return payload_converter - return TemporalIntermediatePayloadConverter(payload_converter) + return _TemporalDataModelPayloadConverter(payload_converter) def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" - intermediate_values: list[Any] = [] + data_model_values: list[Any] = [] for value in values: - to_intermediate = getattr(value, "_temporal_to_intermediate", None) - if to_intermediate is not None: - value = to_intermediate() - intermediate_values.append(value) - return self._inner_payload_converter.to_payloads(intermediate_values) + to_data_model = getattr(value, "_temporal_to_data_model", None) + if to_data_model is not None: + value = to_data_model() + data_model_values.append(value) + return self._inner_payload_converter.to_payloads(data_model_values) def from_payloads( self, @@ -562,7 +562,7 @@ def from_payloads( normalized_type_hints.extend([None] * (len(payloads) - len(type_hints))) inner_type_hints = [ None - if getattr(type_hint, "_temporal_from_intermediate", None) is not None + if getattr(type_hint, "_temporal_from_data_model", None) is not None else type_hint for type_hint in normalized_type_hints ] @@ -570,11 +570,9 @@ def from_payloads( payloads, typing.cast("list[type]", inner_type_hints) ) return [ - from_intermediate(value) + from_data_model(value) if ( - from_intermediate := getattr( - type_hint, "_temporal_from_intermediate", None - ) + from_data_model := getattr(type_hint, "_temporal_from_data_model", None) ) is not None else value diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index e76756cd7..b35a1ce1a 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -11,6 +11,7 @@ import temporalio.converter from temporalio.bridge._visitor_functions import VisitorFunctions from temporalio.converter import BinaryProtoPayloadConverter, CompositePayloadConverter +from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter TEMPORAL_SYSTEM_ENDPOINT = "__temporal_system" _user_payload_converter: contextvars.ContextVar[ @@ -52,13 +53,13 @@ class SystemNexusPayloadConverter(temporalio.converter.PayloadConverter): _user_payload_converter: temporalio.converter.PayloadConverter _outer_payload_converter: temporalio.converter.PayloadConverter - def __init__(self, user_payload_converter: temporalio.converter.PayloadConverter) -> None: + def __init__( + self, user_payload_converter: temporalio.converter.PayloadConverter + ) -> None: """Create a payload converter for system Nexus outer envelopes.""" self._user_payload_converter = user_payload_converter - self._outer_payload_converter = ( - temporalio.converter.TemporalIntermediatePayloadConverter.wrap( - _SystemNexusOuterPayloadConverter() - ) + self._outer_payload_converter = _TemporalDataModelPayloadConverter.wrap( + _SystemNexusOuterPayloadConverter() ) def to_payloads( diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 8e6ba2726..1ed22ec02 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -789,7 +789,7 @@ def _create_workflow_instance( # Create instance from details det = WorkflowInstanceDetails( - payload_converter_class=self._data_converter.payload_converter_class, + payload_converter_factory=self._data_converter._new_payload_converter, failure_converter_class=self._data_converter.failure_converter_class, interceptor_classes=self._interceptor_classes, defn=defn, diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index dd7f4571a..0b4fa0dd0 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -135,7 +135,7 @@ def set_worker_level_failure_exception_types( class WorkflowInstanceDetails: """Immutable details for creating a workflow instance.""" - payload_converter_class: type[temporalio.converter.PayloadConverter] + payload_converter_factory: Callable[[], temporalio.converter.PayloadConverter] failure_converter_class: type[temporalio.converter.FailureConverter] interceptor_classes: Sequence[type[WorkflowInboundInterceptor]] defn: temporalio.workflow._Definition @@ -246,11 +246,7 @@ def __init__(self, det: WorkflowInstanceDetails) -> None: self._defn = det.defn self._workflow_input: ExecuteWorkflowInput | None = None self._info = det.info - self._context_free_payload_converter = ( - temporalio.converter.TemporalIntermediatePayloadConverter.wrap( - det.payload_converter_class() - ) - ) + self._context_free_payload_converter = det.payload_converter_factory() self._context_free_failure_converter = det.failure_converter_class() workflow_context = temporalio.converter.WorkflowSerializationContext( namespace=det.info.namespace, diff --git a/temporalio/worker/workflow_sandbox/_runner.py b/temporalio/worker/workflow_sandbox/_runner.py index b11c9b8c4..3c32c7236 100644 --- a/temporalio/worker/workflow_sandbox/_runner.py +++ b/temporalio/worker/workflow_sandbox/_runner.py @@ -79,7 +79,7 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None: # Just create with fake info which validates self.create_instance( WorkflowInstanceDetails( - payload_converter_class=temporalio.converter.DataConverter.default.payload_converter_class, + payload_converter_factory=temporalio.converter.DataConverter.default._new_payload_converter, failure_converter_class=temporalio.converter.DataConverter.default.failure_converter_class, interceptor_classes=[], defn=defn, diff --git a/tests/test_converter.py b/tests/test_converter.py index 0ae2a2464..6e3a31685 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -44,12 +44,11 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, - PayloadConverter, - TemporalIntermediatePayloadConverter, decode_search_attributes, encode_search_attribute_values, value_to_type, ) +from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter from temporalio.exceptions import ( ApplicationError, FailureError, @@ -261,13 +260,13 @@ class TemporalIntermediateModel: value: str @classmethod - def _temporal_from_intermediate( + def _temporal_from_data_model( cls, - intermediate: temporalio.api.common.v1.WorkflowExecution, + data_model: temporalio.api.common.v1.WorkflowExecution, ) -> TemporalIntermediateModel: - return cls(value=intermediate.workflow_id) + return cls(value=data_model.workflow_id) - def _temporal_to_intermediate(self) -> temporalio.api.common.v1.WorkflowExecution: + def _temporal_to_data_model(self) -> temporalio.api.common.v1.WorkflowExecution: return temporalio.api.common.v1.WorkflowExecution( workflow_id=self.value, run_id="run-id", @@ -278,12 +277,12 @@ class CustomDefaultPayloadConverter(DefaultPayloadConverter): pass -def test_temporal_intermediate_payload_converter_wraps_user_converter(): +def test_temporal_data_model_payload_converter_wraps_user_converter(): data_converter = DataConverter( payload_converter_class=CustomDefaultPayloadConverter ) converter = data_converter.payload_converter - assert isinstance(converter, TemporalIntermediatePayloadConverter) + assert isinstance(converter, _TemporalDataModelPayloadConverter) value = TemporalIntermediateModel("workflow-id") payload = converter.to_payload(value)