Merge branch 'temporalio:main' into main

Author: GSmithApps

.github/workflows/ci.yml

@@ -12,20 +12,13 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python: ["3.8", "3.12"]
+        python: ["3.9", "3.12"]
         os: [ubuntu-latest, macos-intel, macos-arm, windows-latest]
         include:
           - os: macos-intel
             runsOn: macos-13
           - os: macos-arm
             runsOn: macos-14
-        # macOS ARM 3.8 does not have an available Python build at
-        # https://raw.githubusercontent.com/actions/python-versions/main/versions-manifest.json.
-        # See https://github.com/actions/setup-python/issues/808 and
-        # https://github.com/actions/python-versions/pull/259.
-        exclude:
-          - os: macos-arm
-            python: "3.8"
     runs-on: ${{ matrix.runsOn || matrix.os }}
     steps:
       - name: Print build information
@@ -39,11 +32,19 @@ jobs:
       # Using fixed Poetry version until
       # https://github.com/python-poetry/poetry/pull/7694 is fixed
       - run: python -m pip install --upgrade wheel "poetry==1.4.0" poethepoet
-      - run: poetry install --with pydantic --with dsl --with encryption
+      - run: poetry install --with pydantic_converter --with dsl --with encryption --with trio_async
       - run: poe lint
       - run: mkdir junit-xml
-      - run: poe test -s -o log_cli_level=DEBUG --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}.xml
-      - run: poe test -s -o log_cli_level=DEBUG --workflow-environment time-skipping --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}--time-skipping.xml
+      - run: poe test -s --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}.xml
+      - run: poe test -s --workflow-environment time-skipping --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}--time-skipping.xml
+      # This must remain the last step since it downgrades pydantic
+      - name: Uninstall pydantic
+        shell: bash
+        run: |
+          echo y | poetry run pip uninstall pydantic
+          echo y | poetry run pip uninstall pydantic-core
+          poetry run pip install pydantic==1.10
+          poe test -s --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}--pydantic-v1.xml tests/pydantic_converter_v1/workflow_test.py
 
       # On latest, run gevent test
       - name: Gevent test

README.md

@@ -6,7 +6,7 @@ This is the set of Python samples for the [Python SDK](https://github.com/tempor
 
 Prerequisites:
 
-* Python >= 3.8
+* Python >= 3.9
 * [Poetry](https://python-poetry.org)
 * [Temporal CLI installed](https://docs.temporal.io/cli#install)
 * [Local Temporal server running](https://docs.temporal.io/cli/server#start-dev)
@@ -72,6 +72,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models.
 * [schedules](schedules) - Demonstrates a Workflow Execution that occurs according to a schedule.
 * [sentry](sentry) - Report errors to Sentry.
+* [trio_async](trio_async) - Use asyncio Temporal in Trio-based environments.
 * [worker_specific_task_queues](worker_specific_task_queues) - Use unique task queues to ensure activities run on specific workers.
 * [worker_versioning](worker_versioning) - Use the Worker Versioning feature to more easily version your workflows & other code.
 

hello/hello_activity_choice.py

@@ -54,10 +54,10 @@ class ShoppingList:
 @workflow.defn
 class PurchaseFruitsWorkflow:
     @workflow.run
-    async def run(self, list: ShoppingList) -> str:
+    async def run(self, shopping_list: ShoppingList) -> str:
         # Order each thing on the list
         ordered: List[str] = []
-        for item in list.items:
+        for item in shopping_list.items:
             if item.fruit is Fruit.APPLE:
                 order_function = order_apples
             elif item.fruit is Fruit.BANANA:

message_passing/waiting_for_handlers/README.md

@@ -0,0 +1,41 @@
+# Waiting for message handlers
+
+This workflow demonstrates how to wait for signal and update handlers to
+finish in the following circumstances:
+
+- Before a successful return
+- On failure
+- On cancellation
+
+Your workflow can also exit via Continue-As-New. In that case you would
+usually wait for the handlers to finish immediately before the call to
+continue_as_new(); that's not illustrated in this sample.
+
+
+To run, open two terminals and `cd` to this directory in them.
+
+Run the worker in one terminal:
+
+    poetry run python worker.py
+
+And run the workflow-starter code in the other terminal:
+
+    poetry run python starter.py
+
+
+Here's the output you'll see:
+
+```
+workflow exit type: SUCCESS
+    🟢 caller received update result
+    🟢 caller received workflow result
+
+
+workflow exit type: FAILURE
+    🟢 caller received update result
+    🔴 caught exception while waiting for workflow result: Workflow execution failed: deliberately failing workflow
+
+
+workflow exit type: CANCELLATION
+    🟢 caller received update result
+```

message_passing/waiting_for_handlers/__init__.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from enum import IntEnum
+
+TASK_QUEUE = "my-task-queue"
+WORKFLOW_ID = "my-workflow-id"
+
+
+class WorkflowExitType(IntEnum):
+    SUCCESS = 0
+    FAILURE = 1
+    CANCELLATION = 2
+
+
+@dataclass
+class WorkflowInput:
+    exit_type: WorkflowExitType
+
+
+@dataclass
+class WorkflowResult:
+    data: str

message_passing/waiting_for_handlers/activities.py

@@ -0,0 +1,8 @@
+import asyncio
+
+from temporalio import activity
+
+
+@activity.defn
+async def activity_executed_by_update_handler():
+    await asyncio.sleep(1)

message_passing/waiting_for_handlers/starter.py

@@ -0,0 +1,69 @@
+import asyncio
+
+from temporalio import client, common
+
+from message_passing.waiting_for_handlers import (
+    TASK_QUEUE,
+    WORKFLOW_ID,
+    WorkflowExitType,
+    WorkflowInput,
+)
+from message_passing.waiting_for_handlers.workflows import WaitingForHandlersWorkflow
+
+
+async def starter(exit_type: WorkflowExitType):
+    cl = await client.Client.connect("localhost:7233")
+    wf_handle = await cl.start_workflow(
+        WaitingForHandlersWorkflow.run,
+        WorkflowInput(exit_type=exit_type),
+        id=WORKFLOW_ID,
+        task_queue=TASK_QUEUE,
+        id_conflict_policy=common.WorkflowIDConflictPolicy.TERMINATE_EXISTING,
+    )
+    await _check_run(wf_handle, exit_type)
+
+
+async def _check_run(
+    wf_handle: client.WorkflowHandle,
+    exit_type: WorkflowExitType,
+):
+    try:
+        up_handle = await wf_handle.start_update(
+            WaitingForHandlersWorkflow.my_update,
+            wait_for_stage=client.WorkflowUpdateStage.ACCEPTED,
+        )
+    except Exception as e:
+        print(f"    🔴 caught exception while starting update: {e}: {e.__cause__ or ''}")
+
+    if exit_type == WorkflowExitType.CANCELLATION:
+        await wf_handle.cancel()
+
+    try:
+        await up_handle.result()
+        print("    🟢 caller received update result")
+    except Exception as e:
+        print(
+            f"    🔴 caught exception while waiting for update result: {e}: {e.__cause__ or ''}"
+        )
+
+    try:
+        await wf_handle.result()
+        print("    🟢 caller received workflow result")
+    except BaseException as e:
+        print(
+            f"    🔴 caught exception while waiting for workflow result: {e}: {e.__cause__ or ''}"
+        )
+
+
+async def main():
+    for exit_type in [
+        WorkflowExitType.SUCCESS,
+        WorkflowExitType.FAILURE,
+        WorkflowExitType.CANCELLATION,
+    ]:
+        print(f"\n\nworkflow exit type: {exit_type.name}")
+        await starter(exit_type)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

message_passing/waiting_for_handlers/worker.py

@@ -0,0 +1,40 @@
+import asyncio
+import logging
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from message_passing.waiting_for_handlers import TASK_QUEUE
+from message_passing.waiting_for_handlers.activities import (
+    activity_executed_by_update_handler,
+)
+from message_passing.waiting_for_handlers.workflows import WaitingForHandlersWorkflow
+
+interrupt_event = asyncio.Event()
+
+
+async def main():
+    logging.basicConfig(level=logging.INFO)
+
+    client = await Client.connect("localhost:7233")
+
+    async with Worker(
+        client,
+        task_queue=TASK_QUEUE,
+        workflows=[WaitingForHandlersWorkflow],
+        activities=[
+            activity_executed_by_update_handler,
+        ],
+    ):
+        logging.info("Worker started, ctrl+c to exit")
+        await interrupt_event.wait()
+        logging.info("Shutting down")
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())

message_passing/waiting_for_handlers/workflows.py

@@ -0,0 +1,95 @@
+import asyncio
+from datetime import timedelta
+
+from temporalio import exceptions, workflow
+
+from message_passing.waiting_for_handlers import (
+    WorkflowExitType,
+    WorkflowInput,
+    WorkflowResult,
+)
+from message_passing.waiting_for_handlers.activities import (
+    activity_executed_by_update_handler,
+)
+
+
+def is_workflow_exit_exception(e: BaseException) -> bool:
+    """
+    True if the exception is of a type that will cause the workflow to exit.
+
+    This is as opposed to exceptions that cause a workflow task failure, which
+    are retried automatically by Temporal.
+    """
+    # 👉 If you have set additional failure_exception_types you should also
+    # check for these here.
+    return isinstance(e, (asyncio.CancelledError, exceptions.FailureError))
+
+
+@workflow.defn
+class WaitingForHandlersWorkflow:
+    @workflow.run
+    async def run(self, input: WorkflowInput) -> WorkflowResult:
+        """
+        This workflow.run method demonstrates a pattern that can be used to wait for signal and
+        update handlers to finish in the following circumstances:
+
+        - On successful workflow return
+        - On workflow cancellation
+        - On workflow failure
+
+        Your workflow can also exit via Continue-As-New. In that case you would usually wait for
+        the handlers to finish immediately before the call to continue_as_new(); that's not
+        illustrated in this sample.
+
+        If you additionally need to perform cleanup or compensation on workflow failure or
+        cancellation, see the message_passing/waiting_for_handlers_and_compensation sample.
+        """
+        try:
+            # 👉 Use this `try...except` style, instead of waiting for message
+            # handlers to finish in a `finally` block. The reason is that some
+            # exception types cause a workflow task failure as opposed to
+            # workflow exit, in which case we do *not* want to wait for message
+            # handlers to finish.
+            result = await self._my_workflow_application_logic(input)
+            await workflow.wait_condition(workflow.all_handlers_finished)
+            return result
+        # 👉 Catch BaseException since asyncio.CancelledError does not inherit
+        # from Exception.
+        except BaseException as e:
+            if is_workflow_exit_exception(e):
+                await workflow.wait_condition(workflow.all_handlers_finished)
+            raise
+
+    # Methods below this point can be ignored unless you are interested in
+    # the implementation details of this sample.
+
+    def __init__(self) -> None:
+        self._update_started = False
+
+    @workflow.update
+    async def my_update(self) -> str:
+        self._update_started = True
+        await workflow.execute_activity(
+            activity_executed_by_update_handler,
+            start_to_close_timeout=timedelta(seconds=10),
+        )
+        return "update-result"
+
+    async def _my_workflow_application_logic(
+        self, input: WorkflowInput
+    ) -> WorkflowResult:
+        # The main workflow logic is implemented in a separate method in order
+        # to separate "platform-level" concerns (waiting for handlers to finish
+        # and error handling) from application logic.
+
+        # Wait until handlers have started, so that we are demonstrating that we
+        # wait for them to finish.
+        await workflow.wait_condition(lambda: self._update_started)
+        if input.exit_type == WorkflowExitType.SUCCESS:
+            return WorkflowResult(data="workflow-result")
+        elif input.exit_type == WorkflowExitType.FAILURE:
+            raise exceptions.ApplicationError("deliberately failing workflow")
+        elif input.exit_type == WorkflowExitType.CANCELLATION:
+            # Block forever; the starter will send a workflow cancellation request.
+            await asyncio.Future()
+        raise AssertionError("unreachable")

message_passing/waiting_for_handlers_and_compensation/README.md

@@ -0,0 +1,39 @@
+# Waiting for message handlers, and performing compensation and cleanup in message handlers
+
+This sample demonstrates how to do the following:
+
+1. Ensure that all update/signal handlers are finished before a successful
+    workflow return, and on workflow cancellation and failure.
+2. Perform compensation/cleanup in an update handler when the workflow is
+    cancelled or fails.
+
+For a simpler sample showing how to do (1) without (2), see [safe_message_handlers](../safe_message_handlers/README.md).
+
+To run, open two terminals and `cd` to this directory in them.
+
+Run the worker in one terminal:
+
+    poetry run python worker.py
+
+And run the workflow-starter code in the other terminal:
+
+    poetry run python starter.py
+
+
+Here's the output you'll see:
+
+```
+workflow exit type: SUCCESS
+    🟢 caller received update result
+    🟢 caller received workflow result
+
+
+workflow exit type: FAILURE
+    🔴 caught exception while waiting for update result: Workflow update failed: The update failed because the workflow run exited
+    🔴 caught exception while waiting for workflow result: Workflow execution failed: deliberately failing workflow
+
+
+workflow exit type: CANCELLATION
+    🔴 caught exception while waiting for update result: Workflow update failed: The update failed because the workflow run exited
+    🔴 caught exception while waiting for workflow result: Workflow execution failed: Workflow cancelled
+```