Cleanup

Author: tconley1428

scripts/_proto/Dockerfile

@@ -10,7 +10,6 @@ COPY ./ ./
 RUN mkdir -p ./temporalio/api
 RUN uv add "protobuf<4"
 RUN uv sync --all-extras
-RUN poe build-develop
 RUN poe gen-protos
 
 CMD cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new

temporalio/bridge/worker.py

@@ -20,7 +20,6 @@
 )
 
 import google.protobuf.internal.containers
-from google.protobuf.message import Message
 from typing_extensions import TypeAlias
 
 import temporalio.api.common.v1
@@ -278,94 +277,6 @@ async def finalize_shutdown(self) -> None:
         await ref.finalize_shutdown()
 
 
-# See https://mypy.readthedocs.io/en/stable/runtime_troubles.html#using-classes-that-are-generic-in-stubs-but-not-at-runtime
-if TYPE_CHECKING:
-    PayloadContainer: TypeAlias = (
-        google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
-            temporalio.api.common.v1.Payload
-        ]
-    )
-else:
-    PayloadContainer: TypeAlias = (
-        google.protobuf.internal.containers.RepeatedCompositeFieldContainer
-    )
-
-
-async def _apply_to_headers(
-    headers: Mapping[str, temporalio.api.common.v1.Payload],
-    cb: Callable[
-        [Sequence[temporalio.api.common.v1.Payload]],
-        Awaitable[List[temporalio.api.common.v1.Payload]],
-    ],
-) -> None:
-    """Apply API payload callback to headers."""
-    for payload in headers.values():
-        new_payload = (await cb([payload]))[0]
-        payload.CopyFrom(new_payload)
-
-
-async def _decode_headers(
-    headers: Mapping[str, temporalio.api.common.v1.Payload],
-    codec: temporalio.converter.PayloadCodec,
-) -> None:
-    """Decode headers with the given codec."""
-    return await _apply_to_headers(headers, codec.decode)
-
-
-async def _encode_headers(
-    headers: Mapping[str, temporalio.api.common.v1.Payload],
-    codec: temporalio.converter.PayloadCodec,
-) -> None:
-    """Encode headers with the given codec."""
-    return await _apply_to_headers(headers, codec.encode)
-
-
-async def _apply_to_payloads(
-    payloads: PayloadContainer,
-    cb: Callable[
-        [Sequence[temporalio.api.common.v1.Payload]],
-        Awaitable[List[temporalio.api.common.v1.Payload]],
-    ],
-) -> None:
-    """Apply API payload callback to payloads."""
-    if len(payloads) == 0:
-        return
-    new_payloads = await cb(payloads)
-    if new_payloads is payloads:
-        return
-    del payloads[:]
-    # TODO(cretz): Copy too expensive?
-    payloads.extend(new_payloads)
-
-
-async def _apply_to_payload(
-    payload: temporalio.api.common.v1.Payload,
-    cb: Callable[
-        [Sequence[temporalio.api.common.v1.Payload]],
-        Awaitable[List[temporalio.api.common.v1.Payload]],
-    ],
-) -> None:
-    """Apply API payload callback to payload."""
-    new_payload = (await cb([payload]))[0]
-    payload.CopyFrom(new_payload)
-
-
-async def _decode_payloads(
-    payloads: PayloadContainer,
-    codec: temporalio.converter.PayloadCodec,
-) -> None:
-    """Decode payloads with the given codec."""
-    return await _apply_to_payloads(payloads, codec.decode)
-
-
-async def _decode_payload(
-    payload: temporalio.api.common.v1.Payload,
-    codec: temporalio.converter.PayloadCodec,
-) -> None:
-    """Decode a payload with the given codec."""
-    return await _apply_to_payload(payload, codec.decode)
-
-
 async def decode_activation(
     act: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
     codec: temporalio.converter.PayloadCodec,

temporalio/worker/_workflow.py

@@ -563,7 +563,6 @@ def _create_workflow_instance(
             extern_functions=self._extern_functions,
             disable_eager_activity_execution=self._disable_eager_activity_execution,
             worker_level_failure_exception_types=self._workflow_failure_exception_types,
-            last_completion_result=init.last_completion_result,
         )
         if defn.sandboxed:
             return self._workflow_runner.create_instance(det)

temporalio/worker/_workflow_instance.py

@@ -64,7 +64,6 @@
 import temporalio.workflow
 from temporalio.service import __version__
 
-from ..api.common.v1.message_pb2 import Payloads
 from ._interceptor import (
     ContinueAsNewInput,
     ExecuteWorkflowInput,
@@ -144,7 +143,6 @@ class WorkflowInstanceDetails:
     extern_functions: Mapping[str, Callable]
     disable_eager_activity_execution: bool
     worker_level_failure_exception_types: Sequence[Type[BaseException]]
-    last_completion_result: Payloads
 
 
 class WorkflowInstance(ABC):
@@ -322,8 +320,6 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         # metadata query
         self._current_details = ""
 
-        self._last_completion_result = det.last_completion_result
-
         # The versioning behavior of this workflow, as established by annotation or by the dynamic
         # config function. Is only set once upon initialization.
         self._versioning_behavior: Optional[temporalio.common.VersioningBehavior] = None
@@ -1690,34 +1686,6 @@ def workflow_set_current_details(self, details: str):
         self._assert_not_read_only("set current details")
         self._current_details = details
 
-    def workflow_last_completion_result(
-        self, type_hint: Optional[Type]
-    ) -> Optional[Any]:
-        print(
-            "workflow_last_completion_result: ",
-            self._last_completion_result,
-            type(self._last_completion_result),
-            "payload length:",
-            len(self._last_completion_result.payloads),
-        )
-        if len(self._last_completion_result.payloads) == 0:
-            return None
-        elif len(self._last_completion_result.payloads) > 1:
-            warnings.warn(
-                f"Expected single last completion result, got {len(self._last_completion_result.payloads)}"
-            )
-            return None
-
-        print("Payload:", self._last_completion_result.payloads[0])
-        if type_hint is None:
-            return self._payload_converter.from_payload(
-                self._last_completion_result.payloads[0]
-            )
-        else:
-            return self._payload_converter.from_payload(
-                self._last_completion_result.payloads[0], type_hint
-            )
-
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".
 
@@ -2798,7 +2766,6 @@ def _apply_schedule_command(
             v.start_to_close_timeout.FromTimedelta(self._input.start_to_close_timeout)
         if self._input.retry_policy:
             self._input.retry_policy.apply_to_proto(v.retry_policy)
-
         v.cancellation_type = cast(
             temporalio.bridge.proto.workflow_commands.ActivityCancellationType.ValueType,
             int(self._input.cancellation_type),

temporalio/worker/workflow_sandbox/_runner.py

@@ -18,8 +18,6 @@
 import temporalio.worker._workflow_instance
 import temporalio.workflow
 
-from ...api.common.v1.message_pb2 import Payloads
-
 # Workflow instance has to be relative import
 from .._workflow_instance import (
     UnsandboxedWorkflowRunner,
@@ -86,7 +84,6 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
                 extern_functions={},
                 disable_eager_activity_execution=False,
                 worker_level_failure_exception_types=self._worker_level_failure_exception_types,
-                last_completion_result=Payloads(),
             ),
         )
 

temporalio/workflow.py

@@ -897,11 +897,6 @@ def workflow_get_current_details(self) -> str: ...
     @abstractmethod
     def workflow_set_current_details(self, details: str): ...
 
-    @abstractmethod
-    def workflow_last_completion_result(
-        self, type_hint: Optional[Type]
-    ) -> Optional[Any]: ...
-
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(
     "__temporal_current_update_info"

tests/test_client.py

@@ -1,4 +1,3 @@
-import asyncio
 import dataclasses
 import json
 import os

tests/worker/test_workflow.py

@@ -8275,7 +8275,6 @@ async def test_workflow_headers_with_codec(
             "Temporal",
             id=f"workflow-{uuid.uuid4()}",
             task_queue=worker.task_queue,
-            execution_timeout=timedelta(seconds=1),
         )
         assert await workflow_handle.result() == "Hello, Temporal!"
 
@@ -8289,7 +8288,6 @@ async def test_workflow_headers_with_codec(
             SignalAndQueryWorkflow.run,
             id=f"workflow-{uuid.uuid4()}",
             task_queue=worker.task_queue,
-            execution_timeout=timedelta(seconds=1),
         )
 
         # Simple signals and queries
@@ -8329,4 +8327,3 @@ async def test_workflow_headers_with_codec(
             assert headers["foo"].data == b"bar"
         else:
             assert headers["foo"].data != b"bar"
-    assert False