Cleaned things up and got everything functioning

Author: billrich2001

batch_sliding_window/README.md

@@ -12,10 +12,10 @@ A single instance of `SlidingWindowWorkflow` has limited window size and through
 
 To run, first see [README.md](../README.md) for prerequisites. Then, run the following from this directory to start the worker:
 
-    uv run run_worker.py
+    uv run worker.py
 
 This will start the worker. Then, in another terminal, run the following to execute the workflow:
 
-    uv run run_starter.py
+    uv run 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/record_processor_workflow.py

@@ -21,9 +21,8 @@ async def run(self, record: SingleRecord) -> None:
         # 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
-            )
+            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."""

batch_sliding_window/run_starter.py

@@ -39,7 +39,7 @@ class SlidingWindowWorkflow:
 
     def __init__(self):
         self.current_records: Set[int] = set()
-        self.children_started_by_this_run: List[workflow.ChildWorkflowHandle] = []
+        self.children_started_by_this_run = []
         self.offset = 0
         self.progress = 0
         self._completion_signals_received = 0
@@ -117,9 +117,8 @@ async def _continue_as_new_or_complete(self, input: SlidingWindowWorkflowInput)
         new_offset = input.offset + len(self.children_started_by_this_run)
 
         if new_offset < input.maximum_offset:
-            # Wait for all children started by this run to begin execution
-            for child in self.children_started_by_this_run:
-                await child.get_workflow_execution()
+            # In Python, await start_child_workflow() already waits until 
+            # the start has been accepted by the server, so no additional wait needed
 
             # Continue-as-new with updated state
             new_input = SlidingWindowWorkflowInput(

batch_sliding_window/sliding_window_workflow.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+"""Starter for the batch sliding window sample."""
+
+import asyncio
+import datetime
+import logging
+
+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 unique workflow ID with timestamp
+    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    workflow_id = f"batch_sliding_window_example_{timestamp}"
+    
+    # Define workflow input
+    workflow_input = ProcessBatchWorkflowInput(
+        page_size=5,
+        sliding_window_size=10,
+        partitions=3,
+    )
+    
+    print(f"Starting workflow with ID: {workflow_id}")
+    print(f"Input: {workflow_input}")
+    
+    # Start workflow
+    handle = await client.start_workflow(
+        ProcessBatchWorkflow.run,
+        workflow_input,
+        id=workflow_id,
+        task_queue="batch_sliding_window_task_queue",
+    )
+    
+    print(f"Workflow started with ID: {handle.id}")
+    print(f"Waiting for workflow to complete...")
+    
+    # Wait for result
+    try:
+        result = await handle.result()
+        print(f"Workflow completed successfully!")
+        print(f"Total records processed: {result}")
+    except Exception as e:
+        print(f"Workflow failed with error: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

batch_sliding_window/starter.py

@@ -27,7 +27,7 @@ async def main():
     # Create worker
     temporal_worker = worker.Worker(
         client,
-        task_queue="batch_sliding_window",
+        task_queue="batch_sliding_window_task_queue",
         workflows=[
             ProcessBatchWorkflow,
             SlidingWindowWorkflow,