Run linter

Author: billrich2001

batch_sliding_window/__init__.py

@@ -9,7 +9,10 @@
 - RecordLoader: Activity for loading records from external sources
 """
 
-from batch_sliding_window.batch_workflow import ProcessBatchWorkflow, ProcessBatchWorkflowInput
+from batch_sliding_window.batch_workflow import (
+    ProcessBatchWorkflow,
+    ProcessBatchWorkflowInput,
+)
 from batch_sliding_window.sliding_window_workflow import (
     SlidingWindowWorkflow,
     SlidingWindowWorkflowInput,
@@ -34,4 +37,4 @@
     "GetRecordsInput",
     "GetRecordsOutput",
     "SingleRecord",
-] 
+]

batch_sliding_window/batch_workflow.py

@@ -7,16 +7,20 @@
 from temporalio.exceptions import ApplicationError
 
 from batch_sliding_window.record_loader_activity import RecordLoader
-from batch_sliding_window.sliding_window_workflow import SlidingWindowWorkflow, SlidingWindowWorkflowInput
+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 
+
+    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
@@ -25,8 +29,8 @@ class ProcessBatchWorkflowInput:
 @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 
+
+    A real application can choose any other way to divide the records
     into multiple collections.
     """
 
@@ -44,7 +48,9 @@ async def run(self, input: ProcessBatchWorkflowInput) -> int:
             )
 
         partitions = self._divide_into_partitions(record_count, input.partitions)
-        window_sizes = self._divide_into_partitions(input.sliding_window_size, input.partitions)
+        window_sizes = self._divide_into_partitions(
+            input.sliding_window_size, input.partitions
+        )
 
         workflow.logger.info(
             f"ProcessBatchWorkflow started",
@@ -53,22 +59,22 @@ async def run(self, input: ProcessBatchWorkflowInput) -> int:
                 "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],
@@ -77,7 +83,7 @@ async def run(self, input: ProcessBatchWorkflowInput) -> int:
                 progress=0,
                 current_records=None,
             )
-            
+
             task = workflow.execute_child_workflow(
                 SlidingWindowWorkflow.run,
                 child_input,
@@ -100,4 +106,4 @@ def _divide_into_partitions(self, number: int, n: int) -> List[int]:
         for i in range(remainder):
             partitions[i] += 1
 
-        return partitions 
+        return partitions

batch_sliding_window/record_loader_activity.py

@@ -7,6 +7,7 @@
 @dataclass
 class GetRecordsInput:
     """Input for the GetRecords activity."""
+
     page_size: int
     offset: int
     max_offset: int
@@ -15,12 +16,14 @@ class GetRecordsInput:
 @dataclass
 class SingleRecord:
     """Represents a single record to be processed."""
+
     id: int
 
 
 @dataclass
 class GetRecordsOutput:
     """Output from the GetRecords activity."""
+
     records: List[SingleRecord]
 
 
@@ -33,7 +36,7 @@ def __init__(self, record_count: int):
     @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.
         """
@@ -42,13 +45,15 @@ async def get_record_count(self) -> int:
     @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})")
+            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) 
+
+        return GetRecordsOutput(records=records)

batch_sliding_window/record_processor_workflow.py

@@ -13,10 +13,10 @@ class RecordProcessorWorkflow:
     @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:
@@ -29,5 +29,5 @@ async def _process_record(self, record: SingleRecord) -> None:
         # 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}") 
+
+        workflow.logger.info(f"Processed record {record}")

batch_sliding_window/sliding_window_workflow.py

@@ -5,13 +5,18 @@
 from temporalio import workflow
 from temporalio.common import WorkflowIDReusePolicy
 
-from batch_sliding_window.record_loader_activity import RecordLoader, GetRecordsInput, SingleRecord
+from batch_sliding_window.record_loader_activity import (
+    RecordLoader,
+    GetRecordsInput,
+    SingleRecord,
+)
 from batch_sliding_window.record_processor_workflow import RecordProcessorWorkflow
 
 
 @dataclass
 class SlidingWindowWorkflowInput:
     """Contains SlidingWindowWorkflow arguments."""
+
     page_size: int
     sliding_window_size: int
     offset: int  # inclusive
@@ -24,6 +29,7 @@ class SlidingWindowWorkflowInput:
 @dataclass
 class SlidingWindowState:
     """Used as a 'state' query result."""
+
     current_records: List[int]  # record ids currently being processed
     children_started_by_this_run: int
     offset: int
@@ -33,7 +39,7 @@ class SlidingWindowState:
 @workflow.defn
 class SlidingWindowWorkflow:
     """Workflow processes a range of records using a requested number of child workflows.
-    
+
     As soon as a child workflow completes a new one is started.
     """
 
@@ -54,7 +60,7 @@ async def run(self, input: SlidingWindowWorkflowInput) -> int:
                 "offset": input.offset,
                 "maximum_offset": input.maximum_offset,
                 "progress": input.progress,
-            }
+            },
         )
 
         # Initialize state from input
@@ -111,13 +117,15 @@ async def _execute(self, input: SlidingWindowWorkflowInput) -> int:
 
         return await self._continue_as_new_or_complete(input)
 
-    async def _continue_as_new_or_complete(self, input: SlidingWindowWorkflowInput) -> int:
+    async def _continue_as_new_or_complete(
+        self, input: SlidingWindowWorkflowInput
+    ) -> int:
         """Continue-as-new after starting page_size children or complete if done."""
         # Update offset based on children started in this run
         new_offset = input.offset + len(self.children_started_by_this_run)
-        
+
         if new_offset < input.maximum_offset:
-            # In Python, await start_child_workflow() already waits until 
+            # 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
@@ -129,7 +137,7 @@ async def _continue_as_new_or_complete(self, input: SlidingWindowWorkflowInput)
                 progress=self.progress,
                 current_records=self.current_records,
             )
-            
+
             workflow.continue_as_new(new_input)
 
         # Last run in the continue-as-new chain
@@ -152,4 +160,4 @@ def _handle_state_query(self) -> SlidingWindowState:
             children_started_by_this_run=len(self.children_started_by_this_run),
             offset=self.offset,
             progress=self.progress,
-        ) 
+        )

batch_sliding_window/starter.py

@@ -7,7 +7,10 @@
 
 from temporalio.client import Client
 
-from batch_sliding_window.batch_workflow import ProcessBatchWorkflow, ProcessBatchWorkflowInput
+from batch_sliding_window.batch_workflow import (
+    ProcessBatchWorkflow,
+    ProcessBatchWorkflowInput,
+)
 
 
 async def main():
@@ -17,32 +20,32 @@ async def main():
 
     # 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()
@@ -54,4 +57,4 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main()) 
+    asyncio.run(main())

batch_sliding_window/worker.py

@@ -45,4 +45,4 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main()) 
+    asyncio.run(main())