renamed store proxy interfaces

Author: Andrei Neagu

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

@@ -97,7 +97,7 @@ async def start_operation(
         schedule_data_proxy = ScheduleDataStoreProxy(
             store=self._store, schedule_id=schedule_id
         )
-        await schedule_data_proxy.set_multiple(
+        await schedule_data_proxy.create_or_update_multiple(
             {
                 "operation_name": operation_name,
                 "group_index": 0,
@@ -110,7 +110,7 @@ async def start_operation(
             schedule_id=schedule_id,
             operation_name=operation_name,
         )
-        await operation_content_proxy.set_provided_context(initial_operation_context)
+        await operation_content_proxy.create_or_update(initial_operation_context)
 
         await enqueue_schedule_event(self.app, schedule_id)
         return schedule_id
@@ -127,7 +127,7 @@ async def cancel_operation(self, schedule_id: ScheduleId) -> None:
             store=self._store, schedule_id=schedule_id
         )
 
-        is_creating = await schedule_data_proxy.get("is_creating")
+        is_creating = await schedule_data_proxy.read("is_creating")
 
         if is_creating is False:
             _logger.warning(
@@ -136,8 +136,8 @@ async def cancel_operation(self, schedule_id: ScheduleId) -> None:
             )
             return
 
-        operation_name = await schedule_data_proxy.get("operation_name")
-        group_index = await schedule_data_proxy.get("group_index")
+        operation_name = await schedule_data_proxy.read("operation_name")
+        group_index = await schedule_data_proxy.read("group_index")
 
         operation = OperationRegistry.get_operation(operation_name)
         group = operation[group_index]
@@ -172,9 +172,9 @@ async def _cancel_step(step_name: StepName, step_proxy: StepStoreProxy) -> None:
                 f"Cancelling step {step_name=} of {operation_name=} for {schedule_id=}",
             ):
                 with suppress(NoDataFoundError):
-                    deferred_task_uid = await step_proxy.get("deferred_task_uid")
+                    deferred_task_uid = await step_proxy.read("deferred_task_uid")
                     await DeferredRunner.cancel(deferred_task_uid)
-                    await step_proxy.set("status", StepStatus.CANCELLED)
+                    await step_proxy.create_or_update("status", StepStatus.CANCELLED)
 
         await limited_gather(
             *(
@@ -198,9 +198,9 @@ async def restart_operation_step_stuck_in_error(
         schedule_data_proxy = ScheduleDataStoreProxy(
             store=self._store, schedule_id=schedule_id
         )
-        is_creating = await schedule_data_proxy.get("is_creating")
-        operation_name = await schedule_data_proxy.get("operation_name")
-        group_index = await schedule_data_proxy.get("group_index")
+        is_creating = await schedule_data_proxy.read("is_creating")
+        operation_name = await schedule_data_proxy.read("operation_name")
+        group_index = await schedule_data_proxy.read("group_index")
 
         operation = OperationRegistry.get_operation(operation_name)
         step_group = operation[group_index]
@@ -225,7 +225,7 @@ async def restart_operation_step_stuck_in_error(
         )
 
         try:
-            await step_proxy.get("error_traceback")
+            await step_proxy.read("error_traceback")
         except NoDataFoundError as exc:
             raise StepNotInErrorStateError(step_name=step_name) from exc
 
@@ -237,7 +237,7 @@ async def restart_operation_step_stuck_in_error(
         if in_manual_intervention:
             requires_manual_intervention: bool = False
             with suppress(NoDataFoundError):
-                requires_manual_intervention = await step_proxy.get(
+                requires_manual_intervention = await step_proxy.read(
                     "requires_manual_intervention"
                 )
 
@@ -330,9 +330,9 @@ async def _on_schedule_event(self, schedule_id: ScheduleId) -> None:
             store=self._store, schedule_id=schedule_id
         )
 
-        operation_name = await schedule_data_proxy.get("operation_name")
-        is_creating = await schedule_data_proxy.get("is_creating")
-        group_index = await schedule_data_proxy.get("group_index")
+        operation_name = await schedule_data_proxy.read("operation_name")
+        is_creating = await schedule_data_proxy.read("is_creating")
+        group_index = await schedule_data_proxy.read("group_index")
 
         operation = OperationRegistry.get_operation(operation_name)
         step_group = operation[group_index]
@@ -434,7 +434,7 @@ async def _advance_as_repeating(
         steps_stauses = await get_steps_statuses(step_proxies)
         if any(status == StepStatus.CANCELLED for status in steps_stauses.values()):
             # NOTE:
-            await schedule_data_proxy.set("is_creating", value=False)
+            await schedule_data_proxy.create_or_update("is_creating", value=False)
             await enqueue_schedule_event(self.app, schedule_id)
             return
 
@@ -477,7 +477,9 @@ async def _advance_as_creating(
                 next_group_index = group_index + 1
                 # does a next group exist?
                 _ = operation[next_group_index]
-                await schedule_data_proxy.set("group_index", value=next_group_index)
+                await schedule_data_proxy.create_or_update(
+                    "group_index", value=next_group_index
+                )
                 await enqueue_schedule_event(self.app, schedule_id)
             except IndexError:
 
@@ -504,7 +506,9 @@ async def _advance_as_creating(
                     step_name=step.get_step_name(),
                     is_creating=True,
                 )
-                await step_proxy.set("requires_manual_intervention", value=True)
+                await step_proxy.create_or_update(
+                    "requires_manual_intervention", value=True
+                )
                 manual_intervention_step_names.add(step.get_step_name())
 
         if manual_intervention_step_names:
@@ -528,7 +532,7 @@ async def _advance_as_creating(
                 logging.DEBUG,
                 f"{operation_name=} was not successfull: {steps_statuses=}, moving to revert",
             ):
-                await schedule_data_proxy.set("is_creating", value=False)
+                await schedule_data_proxy.create_or_update("is_creating", value=False)
                 await enqueue_schedule_event(self.app, schedule_id)
             return
 
@@ -564,7 +568,9 @@ async def _advance_as_reverting(
                 return
 
             # 1b) -> move to previous group
-            await schedule_data_proxy.set("group_index", value=previous_group_index)
+            await schedule_data_proxy.create_or_update(
+                "group_index", value=previous_group_index
+            )
             await enqueue_schedule_event(self.app, schedule_id)
             return
 

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

@@ -55,7 +55,7 @@ def are_any_steps_in_a_progress_status(
 
 async def _get_step_status(step_proxy: StepStoreProxy) -> tuple[StepName, StepStatus]:
     try:
-        status = await step_proxy.get("status")
+        status = await step_proxy.read("status")
     except NoDataFoundError:
         status = StepStatus.UNKNOWN
 
@@ -86,7 +86,7 @@ async def start_and_mark_as_started(
         is_creating=is_creating,
         expected_steps_count=expected_steps_count,
     )
-    await step_proxy.set_multiple(
+    await step_proxy.create_or_update_multiple(
         {"deferred_created": True, "status": StepStatus.SCHEDULED}
     )
 
@@ -119,7 +119,7 @@ async def get_step_error_traceback(
         step_name=step_name,
         is_creating=False,
     )
-    return step_name, await step_proxy.get("error_traceback")
+    return step_name, await step_proxy.read("error_traceback")
 
 
 def get_group_step_proxies(
@@ -148,7 +148,7 @@ async def _get_was_step_started(
     step_proxy: StepStoreProxy,
 ) -> tuple[bool, StepStoreProxy]:
     try:
-        was_stated = (await step_proxy.get("deferred_created")) is True
+        was_stated = (await step_proxy.read("deferred_created")) is True
     except NoDataFoundError:
         was_stated = False
 
@@ -208,7 +208,7 @@ async def cleanup_after_finishing(
 
 async def get_requires_manual_intervention(step_proxy: StepStoreProxy) -> bool:
     try:
-        return await step_proxy.get("requires_manual_intervention")
+        return await step_proxy.read("requires_manual_intervention")
     except NoDataFoundError:
         return False
 
@@ -220,7 +220,7 @@ async def set_unexpected_opration_state(
     message: str,
 ) -> None:
     schedule_data_proxy = ScheduleDataStoreProxy(store=store, schedule_id=schedule_id)
-    await schedule_data_proxy.set_multiple(
+    await schedule_data_proxy.create_or_update_multiple(
         {
             "operation_error_type": operation_error_type,
             "operation_error_message": message,

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

@@ -158,7 +158,7 @@ async def get_timeout(cls, context: DeferredContext) -> timedelta:
 
     @classmethod
     async def on_created(cls, task_uid: TaskUID, context: DeferredContext) -> None:
-        await get_step_store_proxy(context).set_multiple(
+        await get_step_store_proxy(context).create_or_update_multiple(
             {"deferred_task_uid": task_uid, "status": StepStatus.CREATED}
         )
 
@@ -167,14 +167,16 @@ async def run(cls, context: DeferredContext) -> None:
         app = context["app"]
         is_creating = context["is_creating"]
 
-        await get_step_store_proxy(context).set("status", StepStatus.RUNNING)
+        await get_step_store_proxy(context).create_or_update(
+            "status", StepStatus.RUNNING
+        )
 
         step = _get_step(context)
 
         operation_context_proxy = get_operation_context_proxy(context)
 
         if is_creating:
-            required_context = await operation_context_proxy.get_required_context(
+            required_context = await operation_context_proxy.read(
                 *step.get_create_requires_context_keys()
             )
             _raise_if_any_context_value_is_none(required_context)
@@ -187,7 +189,7 @@ async def run(cls, context: DeferredContext) -> None:
                 provided_operation_context, create_provides_keys
             )
         else:
-            required_context = await operation_context_proxy.get_required_context(
+            required_context = await operation_context_proxy.read(
                 *step.get_revert_requires_context_keys()
             )
             _raise_if_any_context_value_is_none(required_context)
@@ -200,27 +202,31 @@ async def run(cls, context: DeferredContext) -> None:
                 provided_operation_context, revert_provides_keys
             )
 
-        await operation_context_proxy.set_provided_context(provided_operation_context)
+        await operation_context_proxy.create_or_update(provided_operation_context)
 
     @classmethod
     async def on_result(cls, result: None, context: DeferredContext) -> None:
         _ = result
-        await get_step_store_proxy(context).set("status", StepStatus.SUCCESS)
+        await get_step_store_proxy(context).create_or_update(
+            "status", StepStatus.SUCCESS
+        )
 
         await _enqueue_schedule_event_if_group_is_done(context)
 
     @classmethod
     async def on_finished_with_error(
         cls, error: TaskResultError, context: DeferredContext
     ) -> None:
-        await get_step_store_proxy(context).set_multiple(
+        await get_step_store_proxy(context).create_or_update_multiple(
             {"status": StepStatus.FAILED, "error_traceback": error.format_error()}
         )
 
         await _enqueue_schedule_event_if_group_is_done(context)
 
     @classmethod
     async def on_cancelled(cls, context: DeferredContext) -> None:
-        await get_step_store_proxy(context).set("status", StepStatus.CANCELLED)
+        await get_step_store_proxy(context).create_or_update(
+            "status", StepStatus.CANCELLED
+        )
 
         await _enqueue_schedule_event_if_group_is_done(context)