add remap option to rerun_invocation() method

Author: simonbray

bioblend/_tests/TestGalaxyInvocations.py

@@ -145,6 +145,24 @@ def test_rerun_invocation(self):
         history = self.gi.histories.show_history(rerun_invocation['history_id'], contents=True)
         self.assertEqual(len(history), 3)
 
+    @test_util.skip_unless_galaxy('release_21.05')
+    def test_rerun_and_remap_invocation(self):
+        path = test_util.get_abspath(os.path.join('data', 'select_first.ga'))
+        wf = self.gi.workflows.import_workflow_from_local_path(path)
+        wf_inputs = {
+            "0": {'src': 'hda', 'id': self.dataset_id},
+            "1": "-1",
+        }
+        invocation_id = self.gi.workflows.invoke_workflow(wf['id'], inputs=wf_inputs, history_id=self.history_id)['id']
+        errored_jobs = self.gi.jobs.get_jobs(state='error', invocation_id=invocation_id)
+        self.assertEqual(len(errored_jobs), 1)
+        self.gi.invocations.rerun_invocation(invocation_id, remap=True)
+        last_dataset = self.gi.datasets.get_datasets(history_id=self.history_id)[-1]
+        self.gi.datasets.wait_for_dataset(last_dataset['id'])
+        new_errored_jobs = self.gi.jobs.get_jobs(state='error', invocation_id=invocation_id)
+        self.assertEqual(len(new_errored_jobs), 1)
+        self.assertNotEqual(errored_jobs[0]['id'], new_errored_jobs[0]['id'])
+
     def _invoke_workflow(self):
         dataset = {'src': 'hda', 'id': self.dataset_id}
 

bioblend/galaxy/invocations/__init__.py

@@ -132,7 +132,7 @@ def show_invocation(self, invocation_id):
         url = self._make_url(invocation_id)
         return self._get(url=url)
 
-    def rerun_invocation(self, invocation_id: str, inputs_update: Optional[dict] = None,
+    def rerun_invocation(self, invocation_id: str, remap: bool = False, inputs_update: Optional[dict] = None,
                          params_update: Optional[dict] = None, history_id: Optional[str] = None,
                          history_name: Optional[str] = None, import_inputs_to_history: bool = False,
                          replacement_params: Optional[dict] = None, allow_tool_state_corrections: bool = False,
@@ -144,6 +144,13 @@ def rerun_invocation(self, invocation_id: str, inputs_update: Optional[dict] = N
         :type invocation_id: str
         :param invocation_id: Encoded workflow invocation ID to be rerun
 
+        :type remap: bool
+        :param remap: when ``True``, only failed jobs will be rerun. All other parameters
+          for this method will be ignored. Job output(s) will be remapped onto the dataset(s)
+          created by the original jobs; if other jobs were waiting for these jobs to finish
+          successfully, they will be resumed using the new outputs of the tool runs. When ``False``,
+          an entire new invocation will be created, using the other parameters specified.
+
         :type inputs_update: dict
         :param inputs_update: If different datasets should be used to the original
           invocation, this should contain a mapping of workflow inputs to the new
@@ -196,6 +203,18 @@ def rerun_invocation(self, invocation_id: str, inputs_update: Optional[dict] = N
         .. note::
           This method can only be used with Galaxy ``release_21.01`` or later.
         """
+        if remap:
+            errored_jobs = self.gi.jobs.get_jobs(state='error', invocation_id=invocation_id)
+            remap_failures = 0
+            for job in errored_jobs:
+                try:
+                    self.gi.jobs.rerun_job(job['id'], remap=True)
+                except ValueError:
+                    remap_failures += 1
+            if remap_failures:
+                raise ValueError(f'remap was set to True, but {remap_failures} out of {len(errored_jobs)} errored jobs could not be remapped.')
+        return None
+
         invocation_details = self.show_invocation(invocation_id)
         workflow_id = invocation_details['workflow_id']
         inputs = invocation_details['inputs']