Merge branch 'main' into reuse-policy-terminate-existing

Author: mjameswh

.github/workflows/ci.yml

@@ -1,4 +1,7 @@
 name: Continuous Integration
+permissions:
+  contents: read
+  actions: write
 on: # rebuild any PRs and main branch changes
   pull_request:
   push:
@@ -12,13 +15,13 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python: ["3.9", "3.12"]
+        python: ["3.10", "3.14"]
         os: [ubuntu-latest, macos-intel, macos-arm, windows-latest]
         include:
           - os: macos-intel
-            runsOn: macos-13
+            runsOn: macos-15-intel
           - os: macos-arm
-            runsOn: macos-14
+            runsOn: macos-latest
     runs-on: ${{ matrix.runsOn || matrix.os }}
     steps:
       - uses: astral-sh/setup-uv@v5

.gitignore

@@ -3,3 +3,4 @@
 __pycache__
 .vscode
 .DS_Store
+.claude

README.md

@@ -10,7 +10,7 @@ Prerequisites:
 * [Temporal CLI installed](https://docs.temporal.io/cli#install)
 * [Local Temporal server running](https://docs.temporal.io/cli/server#start-dev)
 
-The SDK requires Python >= 3.9. You can install Python using uv. For example,
+The SDK requires Python >= 3.10. You can install Python using uv. For example,
 
     uv python install 3.13
 
@@ -41,6 +41,7 @@ Some examples require extra dependencies. See each sample's directory for specif
   * [hello_async_activity_completion](hello/hello_async_activity_completion.py) - Complete an activity outside of the
     function that was called.
   * [hello_cancellation](hello/hello_cancellation.py) - Manually react to cancellation inside workflows and activities.
+  * [hello_change_log_level](hello/hello_change_log_level.py) - Change the level of workflow task failure from WARN to ERROR.
   * [hello_child_workflow](hello/hello_child_workflow.py) - Execute a child workflow from a workflow.
   * [hello_continue_as_new](hello/hello_continue_as_new.py) - Use continue as new to restart a workflow.
   * [hello_cron](hello/hello_cron.py) - Execute a workflow once a minute.
@@ -56,14 +57,17 @@ Some examples require extra dependencies. See each sample's directory for specif
   * [hello_signal](hello/hello_signal.py) - Send signals to a workflow.
 <!-- Keep this list in alphabetical order -->
 * [activity_worker](activity_worker) - Use Python activities from a workflow in another language.
+* [batch_sliding_window](batch_sliding_window) - Batch processing with a sliding window of child workflows.
 * [bedrock](bedrock) - Orchestrate a chatbot with Amazon Bedrock.
 * [cloud_export_to_parquet](cloud_export_to_parquet) - Set up schedule workflow to process exported files on an hourly basis
 * [context_propagation](context_propagation) - Context propagation through workflows/activities via interceptor.
 * [custom_converter](custom_converter) - Use a custom payload converter to handle custom types.
 * [custom_decorator](custom_decorator) - Custom decorator to auto-heartbeat a long-running activity.
 * [custom_metric](custom_metric) - Custom metric to record the workflow type in the activity schedule to start latency.
 * [dsl](dsl) - DSL workflow that executes steps defined in a YAML file.
+* [eager_wf_start](eager_wf_start) - Run a workflow using Eager Workflow Start
 * [encryption](encryption) - Apply end-to-end encryption for all input/output.
+* [env_config](env_config) - Load client configuration from TOML files with programmatic overrides.
 * [gevent_async](gevent_async) - Combine gevent and Temporal.
 * [langchain](langchain) - Orchestrate workflows for LangChain.
 * [message_passing/introduction](message_passing/introduction/) - Introduction to queries, signals, and updates.
@@ -80,13 +84,10 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [updatable_timer](updatable_timer) - A timer that can be updated while sleeping.
 * [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.
+* [worker_multiprocessing](worker_multiprocessing) - Leverage Python multiprocessing to parallelize workflow tasks and other CPU bound operations by running multiple workers. 
 
 ## Test
 
-Running the tests requires `poe` to be installed.
+To run the tests:
 
-    uv tool install poethepoet
-
-Once you have `poe` installed you can run:
-
-    poe test
+    uv run poe test

activity_worker/README.md

@@ -6,8 +6,8 @@ First run the Go workflow worker by running this in the `go_workflow` directory
 
     go run .
 
-Then in another terminal, run the sample from this directory:
+Then in another terminal, run the sample from the root directory:
 
-    uv run activity_worker.py
+    uv run activity_worker/activity_worker.py
 
 The Python code will invoke the Go workflow which will execute the Python activity and return.

activity_worker/activity_worker.py

@@ -4,6 +4,7 @@
 
 from temporalio import activity
 from temporalio.client import Client
+from temporalio.envconfig import ClientConfig
 from temporalio.worker import Worker
 
 task_queue = "say-hello-task-queue"
@@ -18,7 +19,9 @@ async def say_hello_activity(name: str) -> str:
 
 async def main():
     # Create client to localhost on default namespace
-    client = await Client.connect("localhost:7233")
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
 
     # Run activity worker
     async with Worker(client, task_queue=task_queue, activities=[say_hello_activity]):

batch_sliding_window/README.md

@@ -0,0 +1,21 @@
+# Batch Sliding Window
+
+This sample demonstrates a batch processing workflow that maintains a sliding window of record processing workflows.
+
+A `SlidingWindowWorkflow` starts a configured number (sliding window size) of `RecordProcessorWorkflow` children in parallel. Each child processes a single record. When a child completes, a new child is started.
+
+The `SlidingWindowWorkflow` calls continue-as-new after starting a preconfigured number of children to keep its history size bounded. A `RecordProcessorWorkflow` reports its completion through a signal to its parent, which allows notification of a parent that called continue-as-new.
+
+A single instance of `SlidingWindowWorkflow` has limited window size and throughput. To support larger window size and overall throughput, multiple instances of `SlidingWindowWorkflow` run in parallel.
+
+### Running This Sample
+
+To run, first see [README.md](../README.md) for prerequisites. Then, run the following from root directory to start the worker:
+
+    uv run batch_sliding_window/worker.py
+
+This will start the worker. Then, in another terminal, run the following to execute the workflow:
+
+    uv run batch_sliding_window/starter.py
+
+The workflow will process 90 records using a sliding window of 10 parallel workers across 3 partitions, with a page size of 5 records per continue-as-new iteration.

batch_sliding_window/__init__.py

@@ -0,0 +1,40 @@
+"""Sliding Window Batch Processing Sample.
+
+This sample demonstrates a batch processing workflow that maintains a sliding window
+of record processing workflows. It includes:
+
+- ProcessBatchWorkflow: Main workflow that partitions work across multiple sliding windows
+- SlidingWindowWorkflow: Implements the sliding window pattern with continue-as-new
+- RecordProcessorWorkflow: Processes individual records
+- RecordLoader: Activity for loading records from external sources
+"""
+
+from batch_sliding_window.batch_workflow import (
+    ProcessBatchWorkflow,
+    ProcessBatchWorkflowInput,
+)
+from batch_sliding_window.record_loader_activity import (
+    GetRecordsInput,
+    GetRecordsOutput,
+    RecordLoader,
+    SingleRecord,
+)
+from batch_sliding_window.record_processor_workflow import RecordProcessorWorkflow
+from batch_sliding_window.sliding_window_workflow import (
+    SlidingWindowState,
+    SlidingWindowWorkflow,
+    SlidingWindowWorkflowInput,
+)
+
+__all__ = [
+    "ProcessBatchWorkflow",
+    "ProcessBatchWorkflowInput",
+    "SlidingWindowWorkflow",
+    "SlidingWindowWorkflowInput",
+    "SlidingWindowState",
+    "RecordProcessorWorkflow",
+    "RecordLoader",
+    "GetRecordsInput",
+    "GetRecordsOutput",
+    "SingleRecord",
+]

batch_sliding_window/batch_workflow.py

@@ -0,0 +1,110 @@
+import asyncio
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import List
+
+from temporalio import workflow
+from temporalio.common import WorkflowIDReusePolicy
+from temporalio.exceptions import ApplicationError
+
+from batch_sliding_window.record_loader_activity import RecordLoader
+from batch_sliding_window.sliding_window_workflow import (
+    SlidingWindowWorkflow,
+    SlidingWindowWorkflowInput,
+)
+
+
+@dataclass
+class ProcessBatchWorkflowInput:
+    """Input for the ProcessBatchWorkflow.
+
+    A single input structure is preferred to multiple workflow arguments
+    to simplify backward compatible API changes.
+    """
+
+    page_size: int  # Number of children started by a single sliding window workflow run
+    sliding_window_size: int  # Maximum number of children to run in parallel
+    partitions: int  # How many sliding windows to run in parallel
+
+
+@workflow.defn
+class ProcessBatchWorkflow:
+    """Sample workflow that partitions the data set into continuous ranges.
+
+    A real application can choose any other way to divide the records
+    into multiple collections.
+    """
+
+    @workflow.run
+    async def run(self, input: ProcessBatchWorkflowInput) -> int:
+        # Get total record count
+        record_count: int = await workflow.execute_activity_method(
+            RecordLoader.get_record_count,
+            start_to_close_timeout=timedelta(seconds=5),
+        )
+
+        if input.sliding_window_size < input.partitions:
+            raise ApplicationError(
+                "SlidingWindowSize cannot be less than number of partitions"
+            )
+
+        partitions = self._divide_into_partitions(record_count, input.partitions)
+        window_sizes = self._divide_into_partitions(
+            input.sliding_window_size, input.partitions
+        )
+
+        workflow.logger.info(
+            f"ProcessBatchWorkflow started",
+            extra={
+                "input": input,
+                "record_count": record_count,
+                "partitions": partitions,
+                "window_sizes": window_sizes,
+            },
+        )
+
+        # Start child workflows for each partition
+        tasks = []
+        offset = 0
+
+        for i in range(input.partitions):
+            # Make child id more user-friendly
+            child_id = f"{workflow.info().workflow_id}/{i}"
+
+            # Define partition boundaries
+            maximum_partition_offset = offset + partitions[i]
+            if maximum_partition_offset > record_count:
+                maximum_partition_offset = record_count
+
+            child_input = SlidingWindowWorkflowInput(
+                page_size=input.page_size,
+                sliding_window_size=window_sizes[i],
+                offset=offset,  # inclusive
+                maximum_offset=maximum_partition_offset,  # exclusive
+                progress=0,
+                current_records=None,
+            )
+
+            task = workflow.execute_child_workflow(
+                SlidingWindowWorkflow.run,
+                child_input,
+                id=child_id,
+                id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
+            )
+            tasks.append(task)
+            offset += partitions[i]
+
+        # Wait for all child workflows to complete
+        results = await asyncio.gather(*tasks)
+        return sum(results)
+
+    def _divide_into_partitions(self, number: int, n: int) -> List[int]:
+        """Divide a number into n partitions as evenly as possible."""
+        base = number // n
+        remainder = number % n
+        partitions = [base] * n
+
+        for i in range(remainder):
+            partitions[i] += 1
+
+        return partitions

batch_sliding_window/record_loader_activity.py

@@ -0,0 +1,59 @@
+from dataclasses import dataclass
+from typing import List
+
+from temporalio import activity
+
+
+@dataclass
+class GetRecordsInput:
+    """Input for the GetRecords activity."""
+
+    page_size: int
+    offset: int
+    max_offset: int
+
+
+@dataclass
+class SingleRecord:
+    """Represents a single record to be processed."""
+
+    id: int
+
+
+@dataclass
+class GetRecordsOutput:
+    """Output from the GetRecords activity."""
+
+    records: List[SingleRecord]
+
+
+class RecordLoader:
+    """Activities for loading records from an external data source."""
+
+    def __init__(self, record_count: int):
+        self.record_count = record_count
+
+    @activity.defn
+    async def get_record_count(self) -> int:
+        """Get the total record count.
+
+        Used to partition processing across parallel sliding windows.
+        The sample implementation just returns a fake value passed during worker initialization.
+        """
+        return self.record_count
+
+    @activity.defn
+    async def get_records(self, input: GetRecordsInput) -> GetRecordsOutput:
+        """Get records loaded from an external data source.
+
+        The sample returns fake records.
+        """
+        if input.max_offset > self.record_count:
+            raise ValueError(
+                f"max_offset({input.max_offset}) > record_count({self.record_count})"
+            )
+
+        limit = min(input.offset + input.page_size, input.max_offset)
+        records = [SingleRecord(id=i) for i in range(input.offset, limit)]
+
+        return GetRecordsOutput(records=records)

batch_sliding_window/record_processor_workflow.py

@@ -0,0 +1,33 @@
+import asyncio
+import random
+
+from temporalio import workflow
+
+from batch_sliding_window.record_loader_activity import SingleRecord
+
+
+@workflow.defn
+class RecordProcessorWorkflow:
+    """Workflow that implements processing of a single record."""
+
+    @workflow.run
+    async def run(self, record: SingleRecord) -> None:
+        await self._process_record(record)
+
+        # Notify parent about completion via signal
+        parent = workflow.info().parent
+
+        # This workflow is always expected to have a parent.
+        # But for unit testing it might be useful to skip the notification if there is none.
+        if parent:
+            # Don't specify run_id as parent calls continue-as-new
+            handle = workflow.get_external_workflow_handle(parent.workflow_id)
+            await handle.signal("report_completion", record.id)
+
+    async def _process_record(self, record: SingleRecord) -> None:
+        """Simulate application specific record processing."""
+        # Use workflow.random() to get a random number to ensure workflow determinism
+        sleep_duration = workflow.random().randint(1, 10)
+        await workflow.sleep(sleep_duration)
+
+        workflow.logger.info(f"Processed record {record}")