translated Go sample of sliding window to be writting in python

Author: billrich2001

batch-sliding-window/README.md

@@ -0,0 +1,29 @@
+## Sliding Window Batch Sample 
+
+A sample implementation of 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.
+
+A 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.
+This allows to notify 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 the Sliding Window Batch Sample
+
+Make sure the [Temporal Server is running locally](https://learn.temporal.io/getting_started/python/dev_environment/#set-up-a-local-temporal-service-for-development-with-temporal-cli).
+
+From the root of the project, start a Worker:
+
+```bash
+python python-samples/batch-sliding-window/worker.py
+```
+
+Start the Workflow Execution:
+
+```bash
+python python-samples/batch-sliding-window/starter.py
+``` 

batch-sliding-window/__init__.py

@@ -0,0 +1,37 @@
+"""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_workflow import ProcessBatchWorkflow, ProcessBatchWorkflowInput
+from .sliding_window_workflow import (
+    SlidingWindowWorkflow,
+    SlidingWindowWorkflowInput,
+    SlidingWindowState,
+)
+from .record_processor_workflow import RecordProcessorWorkflow
+from .record_loader_activity import (
+    RecordLoader,
+    GetRecordsInput,
+    GetRecordsOutput,
+    SingleRecord,
+)
+
+__all__ = [
+    "ProcessBatchWorkflow",
+    "ProcessBatchWorkflowInput",
+    "SlidingWindowWorkflow",
+    "SlidingWindowWorkflowInput",
+    "SlidingWindowState",
+    "RecordProcessorWorkflow",
+    "RecordLoader",
+    "GetRecordsInput",
+    "GetRecordsOutput",
+    "SingleRecord",
+] 

batch-sliding-window/batch_workflow.py

@@ -0,0 +1,103 @@
+from dataclasses import dataclass
+from typing import List
+import asyncio
+
+from temporalio import workflow
+from temporalio.common import WorkflowIDReusePolicy
+from temporalio.exceptions import ApplicationError
+
+from .record_loader_activity import RecordLoader
+from .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 = await workflow.execute_activity(
+            RecordLoader.get_record_count,
+            start_to_close_timeout=workflow.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,54 @@
+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,34 @@
+import asyncio
+import random
+
+from temporalio import workflow
+
+from .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
+            await workflow.external_workflow_handle(parent.workflow_id).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}") 

batch-sliding-window/run_starter.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Standalone starter for the batch sliding window sample."""
+
+import asyncio
+import logging
+import sys
+from pathlib import Path
+
+# Add the python-samples directory to the path to enable imports
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from temporalio.client import Client
+
+from batch_sliding_window.batch_workflow import ProcessBatchWorkflow, ProcessBatchWorkflowInput
+
+
+async def main():
+    """Start the ProcessBatchWorkflow."""
+    # Set up logging
+    logging.basicConfig(level=logging.INFO)
+
+    # Create client
+    client = await Client.connect("localhost:7233")
+
+    # Create workflow input
+    workflow_input = ProcessBatchWorkflowInput(
+        page_size=5,
+        sliding_window_size=10,
+        partitions=3,
+    )
+
+    print(f"Starting workflow with input: {workflow_input}")
+
+    # Start workflow
+    handle = await client.start_workflow(
+        ProcessBatchWorkflow.run,
+        workflow_input,
+        id="batch-sliding-window-example",
+        task_queue="batch-sliding-window",
+    )
+
+    print(f"Started workflow: {handle.id}")
+
+    # Wait for workflow completion
+    # This is rarely needed in real use cases as batch workflows are usually long-running
+    result = await handle.result()
+    print(f"Workflow completed. Total records processed: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

batch-sliding-window/run_worker.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Standalone worker runner for the batch sliding window sample."""
+
+import asyncio
+import logging
+import sys
+from pathlib import Path
+
+# Add the python-samples directory to the path to enable imports
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from temporalio import worker
+from temporalio.client import Client
+
+from batch_sliding_window.batch_workflow import ProcessBatchWorkflow
+from batch_sliding_window.sliding_window_workflow import SlidingWindowWorkflow
+from batch_sliding_window.record_processor_workflow import RecordProcessorWorkflow
+from batch_sliding_window.record_loader_activity import RecordLoader
+
+
+async def main():
+    """Run the worker that registers all workflows and activities."""
+    # Set up logging
+    logging.basicConfig(level=logging.INFO)
+
+    # Create client
+    client = await Client.connect("localhost:7233")
+
+    # Create RecordLoader activity with sample data
+    record_loader = RecordLoader(record_count=90)
+
+    # Create worker
+    temporal_worker = worker.Worker(
+        client,
+        task_queue="batch-sliding-window",
+        workflows=[
+            ProcessBatchWorkflow,
+            SlidingWindowWorkflow,
+            RecordProcessorWorkflow,
+        ],
+        activities=[
+            record_loader.get_record_count,
+            record_loader.get_records,
+        ],
+    )
+
+    print("Starting worker...")
+    # Run the worker
+    await temporal_worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())