refactor

Author: Andrei Neagu

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/generic_scheduler/_core.py

@@ -349,7 +349,16 @@ async def _on_schedule_event(self, schedule_id: ScheduleId) -> None:
         _logger.debug("Operation completed: steps_statuses=%s", steps_statuses)
 
         # NOTE: at this point all steps are in a final status
-        if is_creating:
+        if step_group.repeat_steps is True:
+            await self._continue_as_repeat_steps(
+                schedule_data_proxy,
+                schedule_id,
+                operation_name,
+                group_index,
+                step_group,
+                group_step_proxies,
+            )
+        elif is_creating:
             await self._continue_handling_as_creation(
                 steps_statuses,
                 schedule_data_proxy,
@@ -358,7 +367,6 @@ async def _on_schedule_event(self, schedule_id: ScheduleId) -> None:
                 group_index,
                 step_group,
                 operation,
-                group_step_proxies,
             )
         else:
             await self._continue_handling_as_reverting(
@@ -370,60 +378,62 @@ async def _on_schedule_event(self, schedule_id: ScheduleId) -> None:
                 step_group,
             )
 
-    async def _continue_handling_as_creation(
+    async def _continue_as_repeat_steps(
         self,
-        steps_statuses: dict[StepName, StepStatus],
         schedule_data_proxy: ScheduleDataStoreProxy,
         schedule_id: ScheduleId,
         operation_name: OperationName,
         group_index: NonNegativeInt,
         current_step_group: BaseStepGroup,
-        operation: Operation,
         group_step_proxies: dict[StepName, StepStoreProxy],
     ) -> None:
-        # does step require repeating?
-        if current_step_group.repeat_steps is True:
-            # TODO: all this could even be a separate function since the repeat looks like a thing on its own
-            _logger.debug(
-                "REPEATING step_group='%s' in operation_name='%s' for schedule_id='%s'",
-                current_step_group.get_step_group_name(index=group_index),
-                operation_name,
-                schedule_id,
-            )
-            # wait before repeating
-            await asyncio.sleep(current_step_group.wait_before_repeat.total_seconds())
+        _logger.debug(
+            "REPEATING step_group='%s' in operation_name='%s' for schedule_id='%s'",
+            current_step_group.get_step_group_name(index=group_index),
+            operation_name,
+            schedule_id,
+        )
+        # wait before repeating
+        await asyncio.sleep(current_step_group.wait_before_repeat.total_seconds())
 
-            step_proxies: Iterable[StepStoreProxy] = group_step_proxies.values()
+        step_proxies: Iterable[StepStoreProxy] = group_step_proxies.values()
 
-            requeired_steps_statuses = await _get_steps_statuses(step_proxies)
-            if any(
-                status == StepStatus.CANCELLED
-                for status in requeired_steps_statuses.values()
-            ):
-                # was cancelled
-                await schedule_data_proxy.set("is_creating", value=False)
-                await enqueue_schedule_event(self.app, schedule_id)
-                return
+        requeired_steps_statuses = await _get_steps_statuses(step_proxies)
+        if any(
+            status == StepStatus.CANCELLED
+            for status in requeired_steps_statuses.values()
+        ):
+            # was cancelled
+            await schedule_data_proxy.set("is_creating", value=False)
+            await enqueue_schedule_event(self.app, schedule_id)
+            return
 
-            await limited_gather(
-                *(x.remove() for x in step_proxies),
-                limit=_PARALLEL_STATUS_REQUESTS,
-            )
+        await limited_gather(
+            *(x.remove() for x in step_proxies),
+            limit=_PARALLEL_STATUS_REQUESTS,
+        )
 
-            group_proxy = StepGroupProxy(
-                store=self._store,
-                schedule_id=schedule_id,
-                operation_name=operation_name,
-                step_group_name=current_step_group.get_step_group_name(
-                    index=group_index
-                ),
-                is_creating=True,
-            )
-            await group_proxy.remove()
+        group_proxy = StepGroupProxy(
+            store=self._store,
+            schedule_id=schedule_id,
+            operation_name=operation_name,
+            step_group_name=current_step_group.get_step_group_name(index=group_index),
+            is_creating=True,
+        )
+        await group_proxy.remove()
 
-            await enqueue_schedule_event(self.app, schedule_id)
-            return
+        await enqueue_schedule_event(self.app, schedule_id)
 
+    async def _continue_handling_as_creation(
+        self,
+        steps_statuses: dict[StepName, StepStatus],
+        schedule_data_proxy: ScheduleDataStoreProxy,
+        schedule_id: ScheduleId,
+        operation_name: OperationName,
+        group_index: NonNegativeInt,
+        current_step_group: BaseStepGroup,
+        operation: Operation,
+    ) -> None:
         # if all in SUUCESS -> move to next
         if all(status == StepStatus.SUCCESS for status in steps_statuses.values()):
             try: