From e6c8bbdf4627a309e2b5b60bdc1a2a9603c1bbbf Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 1 Oct 2026 09:17:27 -0700 Subject: [PATCH 1/2] Restore wrapped public payload converters --- CHANGELOG.md | 4 + temporalio/activity.py | 13 +- temporalio/converter/_data_converter.py | 30 +- temporalio/converter/_payload_converter.py | 7 - temporalio/nexus/system/__init__.py | 18 +- temporalio/worker/_activity.py | 4 +- temporalio/worker/_nexus.py | 22 +- temporalio/worker/_workflow.py | 8 +- temporalio/worker/_workflow_instance.py | 27 +- temporalio/worker/workflow_sandbox/_runner.py | 2 +- temporalio/workflow/_context.py | 1 + tests/nexus/test_temporal_system_nexus.py | 63 +- tests/test_converter.py | 182 +++++ tests/test_serialization_context.py | 10 +- tests/test_transfer_types.py | 688 ------------------ tests/testing/test_activity.py | 2 - tests/worker/test_activity.py | 2 - tests/worker/test_visitor.py | 17 +- 18 files changed, 273 insertions(+), 827 deletions(-) delete mode 100644 tests/test_transfer_types.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 08b50fdfb..f49cac51c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,10 @@ to include examples, links to docs, or any other relevant information. ### Changed +- Payload converters exposed by data converters and workflow/activity accessors + retain transfer type conversion, so direct use behaves consistently with SDK + serialization. + ### Deprecated ### :boom: Breaking Changes diff --git a/temporalio/activity.py b/temporalio/activity.py index 768c8d108..91decc64f 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -241,9 +241,13 @@ 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 = _TemporalTransferTypePayloadConverter.wrap( + self.payload_converter_class_or_instance + ) else: - self._payload_converter = self.payload_converter_class_or_instance() + self._payload_converter = _TemporalTransferTypePayloadConverter.wrap( + self.payload_converter_class_or_instance() + ) return self._payload_converter @property @@ -456,11 +460,10 @@ def payload_converter() -> temporalio.converter.PayloadConverter: """Get the payload converter for the current activity. The returned converter has :py:class:`temporalio.converter.ActivitySerializationContext` set. + It includes SDK transfer type conversion. This is often used for dynamic activities to convert payloads. """ - return _TemporalTransferTypePayloadConverter.unwrap( - _Context.current().payload_converter - ) + return _Context.current().payload_converter def metric_meter() -> temporalio.common.MetricMeter: diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 540c9bcd6..f924ccf6c 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -73,7 +73,9 @@ class DataConverter(WithSerializationContext): """Class to instantiate for failure conversion.""" payload_converter: PayloadConverter = dataclasses.field(init=False) - """Payload converter created from the :py:attr:`payload_converter_class`.""" + """Payload converter created from :py:attr:`payload_converter_class` and wrapped + with SDK transfer type conversion, including when called directly. + """ failure_converter: FailureConverter = dataclasses.field(init=False) """Failure converter created from the :py:attr:`failure_converter_class`.""" @@ -89,19 +91,11 @@ class DataConverter(WithSerializationContext): """Singleton default data converter.""" def __post_init__(self) -> None: # noqa: D105 - object.__setattr__(self, "payload_converter", self.payload_converter_class()) + object.__setattr__(self, "payload_converter", self._new_payload_converter()) object.__setattr__(self, "failure_converter", self.failure_converter_class()) - def _get_internal_payload_converter(self) -> PayloadConverter: - """Return the configured converter with SDK transfer type hooks enabled.""" - # Reuse the configured instance so its state and any applied serialization - # context are preserved. A new wrapper does not create a new user converter. - return _TemporalTransferTypePayloadConverter.wrap(self.payload_converter) - - def _new_internal_payload_converter(self) -> PayloadConverter: + def _new_payload_converter(self) -> PayloadConverter: """Create a payload converter instance with SDK transfer type hooks enabled.""" - # Unlike _get_internal_payload_converter, instantiate a fresh user converter - # so workflow instances do not share mutable custom converter state. return _TemporalTransferTypePayloadConverter.wrap( self.payload_converter_class() ) @@ -121,7 +115,7 @@ async def encode( same number as values given, but must be at least one and cannot be more than was given. """ - payloads = self._get_internal_payload_converter().to_payloads(values) + payloads = self.payload_converter.to_payloads(values) payloads = await self._encode_payload_sequence(payloads) payloads = await self._external_store_payload_sequence(payloads) return payloads @@ -143,9 +137,7 @@ async def decode( """ payloads = await self._external_retrieve_payload_sequence(payloads) payloads = await self._decode_payload_sequence(payloads) - return self._get_internal_payload_converter().from_payloads( - payloads, type_hints - ) + return self.payload_converter.from_payloads(payloads, type_hints) async def encode_wrapper( self, values: Sequence[Any] @@ -171,9 +163,7 @@ async def encode_failure( self, exception: BaseException, failure: temporalio.api.failure.v1.Failure ) -> None: """Convert and encode failure.""" - self.failure_converter.to_failure( - exception, self._get_internal_payload_converter(), failure - ) + self.failure_converter.to_failure(exception, self.payload_converter, failure) await _apply_to_failure_payloads(failure, self._transform_outbound_payloads) async def decode_failure( @@ -181,9 +171,7 @@ async def decode_failure( ) -> BaseException: """Decode and convert failure.""" await _apply_to_failure_payloads(failure, self._transform_inbound_payloads) - return self.failure_converter.from_failure( - failure, self._get_internal_payload_converter() - ) + return self.failure_converter.from_failure(failure, self.payload_converter) def with_context(self, context: SerializationContext) -> Self: """Return an instance with context set on the component converters.""" diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index a32de0409..d0537c1a5 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -612,13 +612,6 @@ def wrap(payload_converter: PayloadConverter) -> PayloadConverter: return payload_converter return _TemporalTransferTypePayloadConverter(payload_converter) - @staticmethod - def unwrap(payload_converter: PayloadConverter) -> PayloadConverter: - """Remove this wrapper from a payload converter, if present.""" - if isinstance(payload_converter, _TemporalTransferTypePayloadConverter): - return payload_converter._inner_payload_converter - return payload_converter - def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 52b36e915..1f8f7c6d2 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -59,9 +59,7 @@ def _current_user_converters() -> _SystemNexusUserConverters: def _current_user_payload_converter() -> temporalio.converter.PayloadConverter: # pyright: ignore[reportUnusedFunction] """Return the active user payload converter for system Nexus model conversion.""" - return _TemporalTransferTypePayloadConverter.unwrap( - _current_user_converters().payload_converter - ) + return _current_user_converters().payload_converter def _current_user_failure_converter() -> temporalio.converter.FailureConverter: # pyright: ignore[reportUnusedFunction] @@ -98,12 +96,12 @@ class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter): def __init__( self, - internal_payload_converter: temporalio.converter.PayloadConverter, + user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, ) -> None: """Create a payload converter for system Nexus outer envelopes.""" self._user_converters = _SystemNexusUserConverters( - internal_payload_converter, user_failure_converter + user_payload_converter, user_failure_converter ) self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap( _SystemNexusOuterPayloadConverter() @@ -170,14 +168,12 @@ async def maybe_visit_payload( return payload_converter.to_payload(value) -def _get_system_nexus_payload_converter( # pyright: ignore[reportUnusedFunction] - internal_payload_converter: temporalio.converter.PayloadConverter, +def _get_payload_converter( # pyright: ignore[reportUnusedFunction] + user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, ) -> temporalio.converter.PayloadConverter: - """Return the system envelope converter given an internal, wrapped payload converter.""" - return _SystemNexusPayloadConverter( - internal_payload_converter, user_failure_converter - ) + """Return the fixed payload converter for system Nexus outer envelopes.""" + return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) def _get_serialization_context( # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index f72df9cf4..ded3047fc 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -662,7 +662,7 @@ async def _execute_activity( if not running_activity.cancel_thread_raiser else running_activity.cancel_thread_raiser.shielded ), - payload_converter_class_or_instance=data_converter._get_internal_payload_converter(), + payload_converter_class_or_instance=data_converter.payload_converter, runtime_metric_meter=None if sync_non_threaded else self._metric_meter, client=self._client if not running_activity.sync else None, cancellation_details=running_activity.cancellation_details, @@ -847,7 +847,7 @@ async def heartbeat_with_context(*details: Any) -> None: # The payload converter is the already instantiated one for thread # or the picklable class for non-thread payload_converter_class_or_instance = ( - self._worker._data_converter._get_internal_payload_converter() + self._worker._data_converter.payload_converter if isinstance(input.executor, concurrent.futures.ThreadPoolExecutor) else self._worker._data_converter.payload_converter_class ) diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index 38244e99f..a2f4b8ca7 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -323,7 +323,7 @@ async def _handle_cancel_operation_task( ) data_converter.failure_converter.to_failure( handler_error, - data_converter._get_internal_payload_converter(), + data_converter.payload_converter, completion.failure, ) await self._encode_completion(completion, data_converter) @@ -385,7 +385,7 @@ async def _handle_start_operation_task( handler_error = _exception_to_handler_error(err) data_converter.failure_converter.to_failure( handler_error, - data_converter._get_internal_payload_converter(), + data_converter.payload_converter, completion.failure, ) @@ -471,11 +471,7 @@ async def _start_operation( ) ) elif isinstance(result, nexusrpc.handler.StartOperationResultSync): - [payload] = ( - data_converter._get_internal_payload_converter().to_payloads( - [result.value] - ) - ) + [payload] = data_converter.payload_converter.to_payloads([result.value]) return temporalio.api.nexus.v1.StartOperationResponse( sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync( payload=payload, @@ -506,7 +502,7 @@ async def _start_operation( response = temporalio.api.nexus.v1.StartOperationResponse() data_converter.failure_converter.to_failure( new_err, - data_converter._get_internal_payload_converter(), + data_converter.payload_converter, response.failure, ) return response @@ -601,13 +597,11 @@ async def deserialize( ) from err try: - payload_converter = dc._get_internal_payload_converter() + payload_converter = dc.payload_converter if temporalio.nexus.system._is_system_payload(payload): - payload_converter = ( - temporalio.nexus.system._get_system_nexus_payload_converter( - payload_converter, - dc.failure_converter, - ) + payload_converter = temporalio.nexus.system._get_payload_converter( + dc.payload_converter, + dc.failure_converter, ) [input] = payload_converter.from_payloads( [payload], diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 01304d014..f50955c4c 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -452,7 +452,7 @@ async def _handle_activation( try: data_converter.failure_converter.to_failure( err, - data_converter._get_internal_payload_converter(), + data_converter.payload_converter, failure, ) except Exception as inner_err: @@ -468,7 +468,7 @@ async def _handle_activation( try: data_converter.failure_converter.to_failure( err, - data_converter._get_internal_payload_converter(), + data_converter.payload_converter, completion.failed.failure, ) except Exception as inner_err: @@ -728,9 +728,7 @@ def _create_workflow_instance( # Create instance from details det = WorkflowInstanceDetails( - # Sharing the underlying converter could let another workflow's state - # affect conversion and replay, so each instance needs a fresh one. - payload_converter_factory=self._data_converter._new_internal_payload_converter, + 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 5dcc4aa96..0fefc8fa0 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -60,9 +60,6 @@ import temporalio.nexus.system import temporalio.workflow from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo -from temporalio.converter._payload_converter import ( - _TemporalTransferTypePayloadConverter, -) from temporalio.nexus.system.workflow_service._system_nexus_interceptor import ( _start_system_nexus_operation, _SystemNexusWorkflowOutboundInterceptorTerminal, @@ -1063,11 +1060,9 @@ def _apply_resolve_nexus_operation( ) ) if serialization_context is not None: - payload_converter = ( - temporalio.nexus.system._get_system_nexus_payload_converter( - self._payload_converter_with_context(serialization_context), - self._failure_converter_with_context(serialization_context), - ) + payload_converter = temporalio.nexus.system._get_payload_converter( + self._payload_converter_with_context(serialization_context), + self._failure_converter_with_context(serialization_context), ) [output] = self._convert_payloads( [result.completed], @@ -1498,9 +1493,7 @@ def workflow_patch( return use_patch def workflow_payload_converter(self) -> temporalio.converter.PayloadConverter: - return _TemporalTransferTypePayloadConverter.unwrap( - self._workflow_context_payload_converter - ) + return self._workflow_context_payload_converter def workflow_random(self) -> random.Random: self._assert_not_read_only("random") @@ -2247,20 +2240,18 @@ async def operation_handle_fn() -> OutputT: input.operation_name, input.input, ) - internal_payload_converter = self._workflow_context_payload_converter + user_payload_converter = self._workflow_context_payload_converter user_failure_converter = self._workflow_context_failure_converter if serialization_context is not None: - internal_payload_converter = self._payload_converter_with_context( + user_payload_converter = self._payload_converter_with_context( serialization_context ) user_failure_converter = self._failure_converter_with_context( serialization_context ) - payload_converter = ( - temporalio.nexus.system._get_system_nexus_payload_converter( - internal_payload_converter, - user_failure_converter, - ) + payload_converter = temporalio.nexus.system._get_payload_converter( + user_payload_converter, + user_failure_converter, ) failure_converter = user_failure_converter else: diff --git a/temporalio/worker/workflow_sandbox/_runner.py b/temporalio/worker/workflow_sandbox/_runner.py index 3d467c5de..89ce7b951 100644 --- a/temporalio/worker/workflow_sandbox/_runner.py +++ b/temporalio/worker/workflow_sandbox/_runner.py @@ -80,7 +80,7 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None: # Just create with fake info which validates self.create_instance( WorkflowInstanceDetails( - payload_converter_factory=temporalio.converter.DataConverter.default._new_internal_payload_converter, + 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/temporalio/workflow/_context.py b/temporalio/workflow/_context.py index 37928afb9..1dbc341ba 100644 --- a/temporalio/workflow/_context.py +++ b/temporalio/workflow/_context.py @@ -836,6 +836,7 @@ def payload_converter() -> temporalio.converter.PayloadConverter: """Get the payload converter for the current workflow. The returned converter has :py:class:`temporalio.converter.WorkflowSerializationContext` set. + It includes SDK transfer type conversion. This is often used for dynamic workflows/signals/queries to convert payloads. """ diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 9745b0328..49200aeab 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -99,11 +99,10 @@ def from_transfer_type( class _TrackingFailureConverter(temporalio.converter.DefaultFailureConverter): def __init__( - self, - expected_payload_converter_type: type[temporalio.converter.PayloadConverter], + self, expected_payload_converter: temporalio.converter.PayloadConverter ) -> None: super().__init__() - self.expected_payload_converter_type = expected_payload_converter_type + self.expected_payload_converter = expected_payload_converter self.to_failure_calls = 0 self.from_failure_calls = 0 @@ -113,7 +112,7 @@ def to_failure( payload_converter: temporalio.converter.PayloadConverter, failure: temporalio.api.failure.v1.Failure, ) -> None: - assert isinstance(payload_converter, self.expected_payload_converter_type) + assert payload_converter is self.expected_payload_converter self.to_failure_calls += 1 super().to_failure(exception, payload_converter, failure) @@ -122,7 +121,7 @@ def from_failure( failure: temporalio.api.failure.v1.Failure, payload_converter: temporalio.converter.PayloadConverter, ) -> BaseException: - assert isinstance(payload_converter, self.expected_payload_converter_type) + assert payload_converter is self.expected_payload_converter self.from_failure_calls += 1 return super().from_failure(failure, payload_converter) @@ -434,8 +433,8 @@ 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_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -460,8 +459,8 @@ async def test_nexus_payload_serializer_decodes_system_input() -> None: namespace="target-namespace", headers={"test-header": "header-value"}, ) - payload = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -494,8 +493,8 @@ async def test_nexus_payload_serializer_codec_skips_outer_envelope() -> None: signal="test-signal", namespace="target-namespace", ) - payload = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -542,8 +541,8 @@ async def test_schedule_marked_system_nexus_payload_ignores_endpoint() -> None: schedule = completion.successful.commands[0].schedule_nexus_operation data_converter = temporalio.converter.default() - decoded = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + decoded = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ).from_payload(schedule.input) assert isinstance( @@ -569,8 +568,8 @@ async def test_schedule_unmarked_system_nexus_payload_visits_input_as_regular_pa schedule = completion.successful.commands[0].schedule_nexus_operation assert schedule.input.metadata["visited"] == b"true" data_converter = temporalio.converter.default() - decoded = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + decoded = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ).from_payload(schedule.input) assert isinstance( @@ -692,8 +691,8 @@ def _field_is_repeated(field: FieldDescriptor) -> bool: ) def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ) proto_value = _build_proto_sample(message_type) @@ -708,9 +707,9 @@ def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: def test_system_nexus_uses_user_failure_converter() -> None: - payload_converter = temporalio.converter.default()._get_internal_payload_converter() - failure_converter = _TrackingFailureConverter(DefaultPayloadConverter) - system_converter = nexus_system._get_system_nexus_payload_converter( + payload_converter = temporalio.converter.default().payload_converter + failure_converter = _TrackingFailureConverter(payload_converter) + system_converter = nexus_system._get_payload_converter( payload_converter, failure_converter ) @@ -734,7 +733,7 @@ def test_system_nexus_uses_user_failure_converter() -> None: def test_system_nexus_payload_converter_restores_user_context_on_failure() -> None: outer_data_converter = temporalio.converter.default() outer_converters = nexus_system._SystemNexusUserConverters( - outer_data_converter._get_internal_payload_converter(), + outer_data_converter.payload_converter, outer_data_converter.failure_converter, ) inner_data_converter = temporalio.converter.DataConverter() @@ -746,11 +745,11 @@ def to_failure( payload_converter: temporalio.converter.PayloadConverter, failure: temporalio.api.failure.v1.Failure, ) -> None: - assert isinstance(payload_converter, DefaultPayloadConverter) + assert payload_converter is inner_data_converter.payload_converter raise ValueError("conversion failed") - inner_system_converter = nexus_system._get_system_nexus_payload_converter( - inner_data_converter._get_internal_payload_converter(), + inner_system_converter = nexus_system._get_payload_converter( + inner_data_converter.payload_converter, RaisingFailureConverter(), ) @@ -833,24 +832,20 @@ async def test_signal_with_start_uses_target_workflow_serialization_context( captured_contexts: list[SerializationContext | None] = [] system_payload_converter_wrap_count = 0 - original_get_system_nexus_payload_converter = ( - nexus_system._get_system_nexus_payload_converter - ) + original_get_payload_converter = nexus_system._get_payload_converter - def capture_get_system_nexus_payload_converter( - internal_payload_converter: temporalio.converter.PayloadConverter, + def capture_get_payload_converter( + user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, ) -> temporalio.converter.PayloadConverter: nonlocal system_payload_converter_wrap_count system_payload_converter_wrap_count += 1 - return original_get_system_nexus_payload_converter( - internal_payload_converter, user_failure_converter + return original_get_payload_converter( + user_payload_converter, user_failure_converter ) monkeypatch.setattr( - nexus_system, - "_get_system_nexus_payload_converter", - capture_get_system_nexus_payload_converter, + nexus_system, "_get_payload_converter", capture_get_payload_converter ) caller_config = env.client.config() caller_config["data_converter"] = dataclasses.replace( diff --git a/tests/test_converter.py b/tests/test_converter.py index 8444ae0ba..08e511d03 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -6,6 +6,7 @@ import logging import sys import traceback +import typing from collections import deque from collections.abc import Iterable, Mapping, MutableMapping, Sequence from dataclasses import dataclass @@ -14,8 +15,11 @@ from typing import ( Any, Dict, # type:ignore[reportDeprecated] + Generic, Literal, NewType, + TypeVar, + cast, get_args, get_type_hints, ) @@ -44,11 +48,16 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, + TransferTypeConverter, create_payload_validation_error, decode_search_attributes, encode_search_attribute_values, + transfer_type_convertible, value_to_type, ) +from temporalio.converter._payload_converter import ( + _TemporalTransferTypePayloadConverter, +) from temporalio.exceptions import ( ApplicationError, FailureError, @@ -276,6 +285,179 @@ def test_binary_proto(): assert decoded == proto +class TemporalTransferTypeValueConverter( + TransferTypeConverter[ + "TemporalTransferTypeValue", + temporalio.api.common.v1.WorkflowExecution, + ] +): + transfer_type = temporalio.api.common.v1.WorkflowExecution + + def to_transfer_type( + self, value: TemporalTransferTypeValue + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=value.value, + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeValue], + ) -> TemporalTransferTypeValue: + return TemporalTransferTypeValue(value=value.workflow_id) + + +@transfer_type_convertible(TemporalTransferTypeValueConverter) +@dataclass +class TemporalTransferTypeValue: + value: str + + +class TemporalTransferTypeValueWithoutHintConverter( + TransferTypeConverter[ + "TemporalTransferTypeValueWithoutHint", + temporalio.api.common.v1.WorkflowExecution, + ] +): + def to_transfer_type( + self, value: TemporalTransferTypeValueWithoutHint + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=value.value, + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeValueWithoutHint], + ) -> TemporalTransferTypeValueWithoutHint: + return TemporalTransferTypeValueWithoutHint(value=value.workflow_id) + + +@transfer_type_convertible(TemporalTransferTypeValueWithoutHintConverter) +@dataclass +class TemporalTransferTypeValueWithoutHint: + value: str + + +T = TypeVar("T") + + +@dataclass +class TemporalTransferTypeGenericValue(Generic[T]): + value: T + + +class TemporalTransferTypeGenericValueConverter( + TransferTypeConverter[ + TemporalTransferTypeGenericValue[T], + temporalio.api.common.v1.WorkflowExecution, + ] +): + transfer_type = temporalio.api.common.v1.WorkflowExecution + + def to_transfer_type( + self, value: TemporalTransferTypeGenericValue[T] + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=str(value.value), + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeGenericValue[T]], + ) -> TemporalTransferTypeGenericValue[T]: + converted_value: str | int = value.workflow_id + if typing.get_args(type_hint)[0] is int: + converted_value = int(converted_value) + return TemporalTransferTypeGenericValue(value=cast(T, converted_value)) + + +# Register after both classes are defined so the generic type can be resolved. +transfer_type_convertible(TemporalTransferTypeGenericValueConverter)( + TemporalTransferTypeGenericValue +) + + +class CustomDefaultPayloadConverter(DefaultPayloadConverter): + pass + + +def test_temporal_transfer_type_payload_converter_wraps_user_converter(): + data_converter = DataConverter( + payload_converter_class=CustomDefaultPayloadConverter + ) + converter = data_converter.payload_converter + assert isinstance(converter, _TemporalTransferTypePayloadConverter) + value = TemporalTransferTypeValue("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, TemporalTransferTypeValue) == 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_temporal_transfer_type_payload_converter_without_transfer_type_hint(): + converter = DataConverter.default.payload_converter + value = TemporalTransferTypeValueWithoutHint("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 ( + converter.from_payload(payload, TemporalTransferTypeValueWithoutHint) == value + ) + + +@pytest.mark.parametrize( + ("value", "type_hint"), + [ + ( + TemporalTransferTypeGenericValue("workflow-id"), + TemporalTransferTypeGenericValue[str], + ), + ( + TemporalTransferTypeGenericValue(123), + TemporalTransferTypeGenericValue[int], + ), + ], +) +def test_temporal_transfer_type_payload_converter_with_generic_value( + value: TemporalTransferTypeGenericValue[T], + type_hint: type[TemporalTransferTypeGenericValue[T]], +): + converter = DataConverter.default.payload_converter + + payload = converter.to_payload(value) + + assert converter.from_payload(payload, type_hint) == value + + +def test_transfer_type_convertible_rejects_existing_converter(): + with pytest.raises(TypeError, match="already has a transfer type converter"): + transfer_type_convertible(TemporalTransferTypeValueConverter)( + TemporalTransferTypeValue + ) + + 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] diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 9d32f9c48..eed6d2351 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -160,9 +160,7 @@ def __init__(self): @activity.defn async def passthrough_activity(input: TraceData) -> TraceData: - payload_converter = activity.payload_converter() - assert isinstance(payload_converter, SerializationContextCompositePayloadConverter) - payload_converter.to_payload(input) + activity.payload_converter().to_payload(input) activity.heartbeat(input) # Wait for the heartbeat to be processed so that it modifies the data before the activity returns await asyncio.sleep(0.2) @@ -180,11 +178,7 @@ async def run(self, data: TraceData) -> TraceData: class PayloadConversionWorkflow: @workflow.run async def run(self, data: TraceData) -> TraceData: - payload_converter = workflow.payload_converter() - assert isinstance( - payload_converter, SerializationContextCompositePayloadConverter - ) - payload_converter.to_payload(data) + workflow.payload_converter().to_payload(data) data = await workflow.execute_activity( passthrough_activity, data, diff --git a/tests/test_transfer_types.py b/tests/test_transfer_types.py deleted file mode 100644 index 7b7ca33fc..000000000 --- a/tests/test_transfer_types.py +++ /dev/null @@ -1,688 +0,0 @@ -from __future__ import annotations - -import concurrent.futures -import typing -import uuid -from collections.abc import Sequence -from dataclasses import dataclass -from datetime import timedelta -from typing import Any, Generic, TypeVar, cast - -import nexusrpc -import nexusrpc.handler -import pytest - -import temporalio.activity as activity -import temporalio.api.common.v1 -import temporalio.api.failure.v1 -import temporalio.client -import temporalio.common -import temporalio.converter -import temporalio.exceptions -import temporalio.nexus -import temporalio.testing -import temporalio.worker -import temporalio.workflow as workflow -from tests.helpers import new_worker -from tests.helpers.nexus import make_nexus_endpoint_name - - -class TemporalTransferTypeValueConverter( - temporalio.converter.TransferTypeConverter[ - "TemporalTransferTypeValue", - temporalio.api.common.v1.WorkflowExecution, - ] -): - transfer_type = temporalio.api.common.v1.WorkflowExecution - - def to_transfer_type( - self, value: TemporalTransferTypeValue - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=value.value, - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeValue], - ) -> TemporalTransferTypeValue: - return TemporalTransferTypeValue(value=value.workflow_id) - - -@temporalio.converter.transfer_type_convertible(TemporalTransferTypeValueConverter) -@dataclass -class TemporalTransferTypeValue: - value: str - - -class TemporalTransferTypeValueWithoutHintConverter( - temporalio.converter.TransferTypeConverter[ - "TemporalTransferTypeValueWithoutHint", - temporalio.api.common.v1.WorkflowExecution, - ] -): - def to_transfer_type( - self, value: TemporalTransferTypeValueWithoutHint - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=value.value, - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeValueWithoutHint], - ) -> TemporalTransferTypeValueWithoutHint: - return TemporalTransferTypeValueWithoutHint(value=value.workflow_id) - - -@temporalio.converter.transfer_type_convertible( - TemporalTransferTypeValueWithoutHintConverter -) -@dataclass -class TemporalTransferTypeValueWithoutHint: - value: str - - -T = TypeVar("T") - - -@dataclass -class TemporalTransferTypeGenericValue(Generic[T]): - value: T - - -class TemporalTransferTypeGenericValueConverter( - temporalio.converter.TransferTypeConverter[ - TemporalTransferTypeGenericValue[T], - temporalio.api.common.v1.WorkflowExecution, - ] -): - transfer_type = temporalio.api.common.v1.WorkflowExecution - - def to_transfer_type( - self, value: TemporalTransferTypeGenericValue[T] - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=str(value.value), - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeGenericValue[T]], - ) -> TemporalTransferTypeGenericValue[T]: - converted_value: str | int = value.workflow_id - if typing.get_args(type_hint)[0] is int: - converted_value = int(converted_value) - return TemporalTransferTypeGenericValue(value=cast(T, converted_value)) - - -# Register after both classes are defined so the generic type can be resolved. -temporalio.converter.transfer_type_convertible( - TemporalTransferTypeGenericValueConverter -)(TemporalTransferTypeGenericValue) - - -class CustomDefaultPayloadConverter(temporalio.converter.DefaultPayloadConverter): - pass - - -class ContextualTransferPayloadConverter(CustomDefaultPayloadConverter): - def __init__(self) -> None: - super().__init__() - self.context: temporalio.converter.SerializationContext | None = None - self.encoding_contexts: list[ - temporalio.converter.SerializationContext | None - ] = [] - - def with_context( - self, context: temporalio.converter.SerializationContext - ) -> ContextualTransferPayloadConverter: - converter = ContextualTransferPayloadConverter() - converter.context = context - return converter - - def to_payloads( - self, values: Sequence[Any] - ) -> list[temporalio.api.common.v1.Payload]: - self.encoding_contexts.append(self.context) - return super().to_payloads(values) - - -async def test_transfer_type_conversion_preserves_user_converter_context(): - original = temporalio.converter.DataConverter( - payload_converter_class=ContextualTransferPayloadConverter - ) - context = temporalio.converter.WorkflowSerializationContext( - namespace="test", workflow_id="workflow-id" - ) - converter = original.with_context(context) - user_converter = converter.payload_converter - assert isinstance(user_converter, ContextualTransferPayloadConverter) - assert user_converter.context is context - assert user_converter is not original.payload_converter - value = TemporalTransferTypeValue("workflow-id") - - payloads = await converter.encode([value]) - - assert user_converter.encoding_contexts == [context] - assert await converter.decode(payloads, [TemporalTransferTypeValue]) == [value] - assert converter.payload_converter is user_converter - - -async def test_transfer_type_conversion_in_failure_details(): - converter = temporalio.converter.DataConverter( - payload_converter_class=CustomDefaultPayloadConverter - ) - failure = temporalio.api.failure.v1.Failure() - - await converter.encode_failure( - temporalio.exceptions.ApplicationError( - "test failure", TemporalTransferTypeValue("workflow-id") - ), - failure, - ) - converted = await converter.decode_failure(failure) - - assert isinstance(converted, temporalio.exceptions.ApplicationError) - assert converted.details == ( - temporalio.api.common.v1.WorkflowExecution( - workflow_id="workflow-id", run_id="run-id" - ), - ) - assert isinstance(converter.payload_converter, CustomDefaultPayloadConverter) - - -async def test_temporal_transfer_type_payload_converter_wraps_user_converter(): - data_converter = temporalio.converter.DataConverter( - payload_converter_class=CustomDefaultPayloadConverter - ) - assert isinstance(data_converter.payload_converter, CustomDefaultPayloadConverter) - assert isinstance( - temporalio.converter.PayloadConverter.default, - temporalio.converter.DefaultPayloadConverter, - ) - value = TemporalTransferTypeValue("workflow-id") - - [payload] = await data_converter.encode([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 await data_converter.decode([payload], [TemporalTransferTypeValue]) == [ - value - ] - - [plain_proto_payload] = await data_converter.encode( - [temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2")] - ) - assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" - - -async def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): - converter = temporalio.converter.DataConverter.default - value = TemporalTransferTypeValueWithoutHint("workflow-id") - - [payload] = await converter.encode([value]) - - assert payload.metadata["encoding"] == b"json/protobuf" - assert ( - payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" - ) - assert await converter.decode( - [payload], [TemporalTransferTypeValueWithoutHint] - ) == [value] - - -@pytest.mark.parametrize( - ("value", "type_hint"), - [ - ( - TemporalTransferTypeGenericValue("workflow-id"), - TemporalTransferTypeGenericValue[str], - ), - ( - TemporalTransferTypeGenericValue(123), - TemporalTransferTypeGenericValue[int], - ), - ], -) -async def test_temporal_transfer_type_payload_converter_with_generic_value( - value: TemporalTransferTypeGenericValue[T], - type_hint: type[TemporalTransferTypeGenericValue[T]], -): - converter = temporalio.converter.DataConverter.default - - [payload] = await converter.encode([value]) - - assert await converter.decode([payload], [type_hint]) == [value] - - -def test_transfer_type_convertible_rejects_existing_converter(): - with pytest.raises(TypeError, match="already has a transfer type converter"): - temporalio.converter.transfer_type_convertible( - TemporalTransferTypeValueConverter - )(TemporalTransferTypeValue) - - -class TransferValueConverter( - temporalio.converter.TransferTypeConverter["TransferValue", str] -): - transfer_type = str - - def to_transfer_type(self, value: TransferValue) -> str: - return f"transfer:{value.text}" - - def from_transfer_type( - self, value: str, type_hint: type[TransferValue] - ) -> TransferValue: - assert type_hint is TransferValue - assert value.startswith("transfer:") - return TransferValue(value.removeprefix("transfer:")) - - -@temporalio.converter.transfer_type_convertible(TransferValueConverter) -class TransferValue: - # A non-dataclass ensures missing transfer conversion cannot silently fall - # back to the default converter's dataclass JSON support. - def __init__(self, text: str) -> None: - self.text = text - - def append(self, step: str) -> TransferValue: - return TransferValue(f"{self.text}/{step}") - - -def test_transfer_value_requires_transfer_conversion(): - with pytest.raises(TypeError, match="not JSON serializable"): - temporalio.converter.DefaultPayloadConverter().to_payload( - TransferValue("input") - ) - - -@workflow.defn -class TransferWorkflow: - @workflow.run - async def run(self, value: TransferValue, operation: str) -> TransferValue: - if operation == "child": - result = await workflow.execute_child_workflow( - TransferWorkflow.run, - args=[value.append("child-input"), "return"], - id=f"{workflow.info().workflow_id}-child", - ) - return result.append("parent") - if operation == "continue-as-new": - workflow.continue_as_new(args=[value.append("continued"), "return"]) - assert operation == "return" - return value.append("workflow") - - -@pytest.mark.parametrize( - "operation, expected", - [ - ("return", "input/workflow"), - ("child", "input/child-input/workflow/parent"), - ("continue-as-new", "input/continued/workflow"), - ], -) -async def test_transfer_types_workflow( - client: temporalio.client.Client, operation: str, expected: str -): - async with new_worker( - client, TransferWorkflow, workflow_failure_exception_types=[Exception] - ) as worker: - result = await client.execute_workflow( - TransferWorkflow.run, - args=[TransferValue("input"), operation], - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - ) - assert result.text == expected - - -@activity.defn -async def transfer_activity(value: TransferValue) -> TransferValue: - return value.append("activity") - - -@activity.defn -def transfer_sync_activity(value: TransferValue) -> TransferValue: - return value.append("activity") - - -@workflow.defn -class TransferActivityWorkflow: - @workflow.run - async def run( - self, value: TransferValue, local: bool, synchronous: bool - ) -> TransferValue: - activity_fn = transfer_sync_activity if synchronous else transfer_activity - if local: - result = await workflow.execute_local_activity( - activity_fn, - value.append("activity-input"), - start_to_close_timeout=timedelta(seconds=10), - retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), - ) - else: - result = await workflow.execute_activity( - activity_fn, - value.append("activity-input"), - start_to_close_timeout=timedelta(seconds=10), - retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), - ) - assert isinstance(result, TransferValue) - return result.append("workflow") - - -@pytest.mark.parametrize("local", [False, True], ids=["remote", "local"]) -@pytest.mark.parametrize("execution", ["async", "thread", "process"]) -async def test_transfer_types_activity( - client: temporalio.client.Client, - shared_state_manager: temporalio.worker.SharedStateManager, - local: bool, - execution: str, -): - executor_cls = ( - concurrent.futures.ProcessPoolExecutor - if execution == "process" - else concurrent.futures.ThreadPoolExecutor - ) - with executor_cls(max_workers=2) as executor: - async with new_worker( - client, - TransferActivityWorkflow, - activities=[transfer_activity, transfer_sync_activity], - activity_executor=executor, - shared_state_manager=shared_state_manager, - workflow_failure_exception_types=[Exception], - ) as worker: - result = await client.execute_workflow( - TransferActivityWorkflow.run, - args=[TransferValue("input"), local, execution != "async"], - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - ) - assert result.text == "input/activity-input/activity/workflow" - - -@workflow.defn -class TransferMessagesWorkflow: - def __init__(self) -> None: - self.value = TransferValue("unset") - self.finished = False - - @workflow.run - async def run(self, value: TransferValue) -> TransferValue: - await workflow.wait_condition(lambda: self.finished) - return self.value.append(value.text) - - @workflow.signal - def signal(self, value: TransferValue) -> None: - self.value = value.append("signal") - - @workflow.query - def query(self, value: TransferValue) -> TransferValue: - return self.value.append(value.text) - - @workflow.update - async def update(self, value: TransferValue) -> TransferValue: - self.value = value.append("update") - return self.value.append("result") - - @update.validator - def validate_update(self, value: TransferValue) -> None: - assert value.text - - @workflow.signal - def finish(self) -> None: - self.finished = True - - -@pytest.mark.parametrize( - "start", ["workflow", "signal-with-start", "update-with-start"] -) -async def test_transfer_types_messages(client: temporalio.client.Client, start: str): - async with new_worker( - client, TransferMessagesWorkflow, workflow_failure_exception_types=[Exception] - ) as worker: - if start == "update-with-start": - start_op = temporalio.client.WithStartWorkflowOperation( - TransferMessagesWorkflow.run, - TransferValue("workflow-input"), - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - id_conflict_policy=temporalio.common.WorkflowIDConflictPolicy.FAIL, - ) - result = await client.execute_update_with_start_workflow( - TransferMessagesWorkflow.update, - TransferValue("start-update-input"), - start_workflow_operation=start_op, - ) - assert result.text == "start-update-input/update/result" - handle = await start_op.workflow_handle() - else: - handle = await client.start_workflow( - TransferMessagesWorkflow.run, - TransferValue("workflow-input"), - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - start_signal="signal" if start == "signal-with-start" else None, - start_signal_args=[TransferValue("signal-input")] - if start == "signal-with-start" - else [], - ) - if start != "signal-with-start": - await handle.signal( - TransferMessagesWorkflow.signal, TransferValue("signal-input") - ) - result = await handle.query( - TransferMessagesWorkflow.query, TransferValue("query-input") - ) - assert result.text == "signal-input/signal/query-input" - - result = await handle.execute_update( - TransferMessagesWorkflow.update, TransferValue("update-input") - ) - assert result.text == "update-input/update/result" - await handle.signal(TransferMessagesWorkflow.finish) - assert (await handle.result()).text == "update-input/update/workflow-input" - - -@workflow.defn -class TransferSignalWorkflow: - @workflow.run - async def run(self, value: TransferValue, external: bool) -> TransferValue: - child = await workflow.start_child_workflow( - TransferMessagesWorkflow.run, - value.append("child-input"), - id=f"{workflow.info().workflow_id}-child", - ) - if external: - handle: workflow.ExternalWorkflowHandle[TransferMessagesWorkflow] = ( - workflow.get_external_workflow_handle_for( - TransferMessagesWorkflow.run, child.id - ) - ) - await handle.signal( - TransferMessagesWorkflow.signal, value.append("external") - ) - await handle.signal(TransferMessagesWorkflow.finish) - else: - await child.signal(TransferMessagesWorkflow.signal, value.append("child")) - await child.signal(TransferMessagesWorkflow.finish) - return (await child).append("parent") - - -@pytest.mark.parametrize("external", [False, True], ids=["child", "external"]) -async def test_transfer_types_workflow_signal( - client: temporalio.client.Client, external: bool -): - async with new_worker( - client, - TransferSignalWorkflow, - TransferMessagesWorkflow, - workflow_failure_exception_types=[Exception], - ) as worker: - result = await client.execute_workflow( - TransferSignalWorkflow.run, - args=[TransferValue("input"), external], - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - ) - signal_target = "external" if external else "child" - assert result.text == f"input/{signal_target}/signal/input/child-input/parent" - - -@activity.defn -async def transfer_failure_activity(value: TransferValue) -> None: - raise temporalio.exceptions.ApplicationError("failure", value.append("activity")) - - -@workflow.defn -class TransferFailureWorkflow: - @workflow.run - async def run(self, value: TransferValue, operation: str) -> None: - if operation == "workflow": - raise temporalio.exceptions.ApplicationError( - "failure", value.append("workflow") - ) - try: - if operation == "local-activity": - await workflow.execute_local_activity( - transfer_failure_activity, - value, - start_to_close_timeout=timedelta(seconds=10), - retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), - ) - else: - await workflow.execute_activity( - transfer_failure_activity, - value, - start_to_close_timeout=timedelta(seconds=10), - retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), - ) - except temporalio.exceptions.FailureError as err: - cause = ( - err.cause - if isinstance(err, temporalio.exceptions.ActivityError) - else err - ) - assert isinstance(cause, temporalio.exceptions.ApplicationError) - # Failure details have no type hints, so decoding yields the transfer type. - assert cause.details == ("transfer:input/activity",) - raise - - -@pytest.mark.parametrize("operation", ["workflow", "activity", "local-activity"]) -async def test_transfer_types_failure_details( - client: temporalio.client.Client, operation: str -): - async with new_worker( - client, - TransferFailureWorkflow, - activities=[transfer_failure_activity], - workflow_failure_exception_types=[Exception], - ) as worker: - with pytest.raises(temporalio.client.WorkflowFailureError) as err: - await client.execute_workflow( - TransferFailureWorkflow.run, - args=[TransferValue("input"), operation], - id=str(uuid.uuid4()), - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=30), - ) - cause: BaseException | None = err.value.cause - if isinstance(cause, temporalio.exceptions.ActivityError): - cause = cause.cause - assert isinstance(cause, temporalio.exceptions.ApplicationError) - source = "workflow" if operation == "workflow" else "activity" - assert cause.details == (f"transfer:input/{source}",) - - -@nexusrpc.service -class TransferService: - sync: nexusrpc.Operation[TransferValue, TransferValue] - async_: nexusrpc.Operation[TransferValue, TransferValue] - - -@nexusrpc.handler.service_handler(service=TransferService) -class TransferServiceHandler: - @nexusrpc.handler.sync_operation - async def sync( - self, _ctx: nexusrpc.handler.StartOperationContext, input: TransferValue - ) -> TransferValue: - return input.append("nexus-sync") - - @temporalio.nexus.workflow_run_operation - async def async_( - self, ctx: temporalio.nexus.WorkflowRunOperationContext, input: TransferValue - ) -> temporalio.nexus.WorkflowHandle[TransferValue]: - return await ctx.start_workflow( - TransferWorkflow.run, - args=[input.append("nexus-async"), "return"], - id=str(uuid.uuid4()), - ) - - -@workflow.defn -class TransferNexusWorkflow: - @workflow.run - async def run( - self, value: TransferValue, endpoint: str, synchronous: bool - ) -> TransferValue: - client = workflow.create_nexus_client( - service=TransferService, endpoint=endpoint - ) - result = await client.execute_operation( - TransferService.sync if synchronous else TransferService.async_, - value.append("nexus-input"), - schedule_to_close_timeout=timedelta(seconds=20), - ) - return result.append("caller") - - -# Cloud CI credentials cannot manage Nexus endpoints. -@pytest.mark.requires_local_server -@pytest.mark.parametrize("synchronous", [False, True], ids=["async", "sync"]) -async def test_transfer_types_nexus( - env: temporalio.testing.WorkflowEnvironment, synchronous: bool -): - if env.supports_time_skipping: - pytest.skip("Nexus requires the dev server") - task_queue = str(uuid.uuid4()) - endpoint = await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) - try: - async with new_worker( - env.client, - TransferNexusWorkflow, - TransferWorkflow, - task_queue=task_queue, - nexus_service_handlers=[TransferServiceHandler()], - workflow_failure_exception_types=[Exception], - ): - result = await env.client.execute_workflow( - TransferNexusWorkflow.run, - args=[TransferValue("input"), endpoint.spec.name, synchronous], - id=str(uuid.uuid4()), - task_queue=task_queue, - execution_timeout=timedelta(seconds=30), - ) - expected = "nexus-sync" if synchronous else "nexus-async/workflow" - assert result.text == f"input/nexus-input/{expected}/caller" - finally: - await env.delete_nexus_endpoint(endpoint) diff --git a/tests/testing/test_activity.py b/tests/testing/test_activity.py index 74e14b99f..71ba7f590 100644 --- a/tests/testing/test_activity.py +++ b/tests/testing/test_activity.py @@ -16,7 +16,6 @@ async def test_activity_env_async(): waiting = asyncio.Event() async def do_stuff(param: str) -> str: - assert activity.payload_converter() is env.payload_converter activity.heartbeat(f"param: {param}") # Ensure it works across create_task @@ -58,7 +57,6 @@ def test_activity_env_sync(): properly_cancelled = False def do_stuff(param: str) -> None: - assert activity.payload_converter() is env.payload_converter activity.heartbeat(f"param: {param}") # Ensure it works across thread diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py index c2add4769..64691a93f 100644 --- a/tests/worker/test_activity.py +++ b/tests/worker/test_activity.py @@ -30,7 +30,6 @@ WorkflowHandle, ) from temporalio.common import RawValue, RetryPolicy -from temporalio.converter import DefaultPayloadConverter from temporalio.exceptions import ( ActivityError, ApplicationError, @@ -1617,7 +1616,6 @@ class DynActivityValue: @activity.defn(dynamic=True) def sync_dyn_activity(args: Sequence[RawValue]) -> DynActivityValue: assert len(args) == 2 - assert isinstance(activity.payload_converter(), DefaultPayloadConverter) arg1 = activity.payload_converter().from_payload(args[0].payload, DynActivityValue) assert isinstance(arg1, DynActivityValue) arg2 = activity.payload_converter().from_payload(args[1].payload, DynActivityValue) diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 320f537aa..362288f34 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -235,8 +235,8 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: input=Payloads(payloads=[Payload(data=b"workflow-input")]), ) data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ) system_payload = payload_converter.to_payload(system_request) @@ -280,8 +280,8 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: run_id="test-run-id" ) data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ) system_payload = payload_converter.to_payload(response) @@ -307,8 +307,8 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: async def test_unknown_system_nexus_payload_raises_application_error(): data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter, ) system_payload = payload_converter.to_payload( @@ -482,9 +482,8 @@ async def _visit(self) -> None: active_visits -= 1 data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_system_nexus_payload_converter( - data_converter._get_internal_payload_converter(), - data_converter.failure_converter, + payload_converter = nexus_system._get_payload_converter( + data_converter.payload_converter, data_converter.failure_converter ) system_request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest( input=Payloads(payloads=[Payload(data=b"workflow-input")]), From 04b68ccb065962964c93629744b758a28ca8d608 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 1 Oct 2026 09:42:01 -0700 Subject: [PATCH 2/2] Trim converter accessor documentation --- temporalio/activity.py | 1 - temporalio/workflow/_context.py | 1 - 2 files changed, 2 deletions(-) diff --git a/temporalio/activity.py b/temporalio/activity.py index 91decc64f..3f69bc17f 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -460,7 +460,6 @@ def payload_converter() -> temporalio.converter.PayloadConverter: """Get the payload converter for the current activity. The returned converter has :py:class:`temporalio.converter.ActivitySerializationContext` set. - It includes SDK transfer type conversion. This is often used for dynamic activities to convert payloads. """ return _Context.current().payload_converter diff --git a/temporalio/workflow/_context.py b/temporalio/workflow/_context.py index 1dbc341ba..37928afb9 100644 --- a/temporalio/workflow/_context.py +++ b/temporalio/workflow/_context.py @@ -836,7 +836,6 @@ def payload_converter() -> temporalio.converter.PayloadConverter: """Get the payload converter for the current workflow. The returned converter has :py:class:`temporalio.converter.WorkflowSerializationContext` set. - It includes SDK transfer type conversion. This is often used for dynamic workflows/signals/queries to convert payloads. """