feat: ✨ Update run log status for Notebook/function execution (#95)

* feat: ✨ Update run log status for Notebook/function execution

* feat: ✨ Update run log status for Notebook/function execution

magnus/datastore.py

@@ -345,6 +345,19 @@ def put_run_log(self, run_log: RunLog, **kwargs):
         """
         raise NotImplementedError
 
+    def update_run_log_status(self, run_id: str, status: str):
+        """
+        Updates the status of the Run Log defined by the run_id
+
+        Args:
+            run_id (str): The run_id of the run
+            status (str): The new status of the run
+        """
+        logger.info(f'Updating status of run_id {run_id} to {status}')
+        run_log = self.get_run_log_by_id(run_id, full=False)
+        run_log.status = status
+        self.put_run_log(run_log)
+
     def get_parameters(self, run_id: str, **kwargs) -> dict:  # pylint: disable=unused-argument
         """
         Get the parameters from the Run log defined by the run_id

magnus/executor.py

@@ -672,6 +672,10 @@ def execute_job(self, node: BaseNode):
         self.run_log_store.add_step_log(step_log, self.run_id)
         self.execute_node(node=node)
 
+        # Update the run log status
+        step_log = self.run_log_store.get_step_log(node._get_step_log_name(), self.run_id)
+        self.run_log_store.update_run_log_status(run_id=self.run_id, status=step_log.status)
+
 
 class LocalContainerExecutor(BaseExecutor):
     """

magnus/pipeline.py

@@ -460,8 +460,11 @@ def execute_nb_or_func(run_id, command: str, catalog_config: dict,  configuratio
 
     # Prepare for graph execution
     mode_executor.prepare_for_node_execution()
-
     logger.info('Executing the job')
     mode_executor.execute_node(node=node)
 
+    # Update the status of the run log
+    step_log = mode_executor.run_log_store.get_step_log(node._get_step_log_name(), run_id)
+    mode_executor.run_log_store.update_run_log_status(run_id=run_id, status=step_log.status)
+
     mode_executor.send_return_code()