Merge pull request #521 from martenson/clean-refactor

Refactor rerun_invocation to use /request as the source of arguments.

Author: nsoranzo

bioblend/_tests/TestGalaxyInvocations.py

@@ -14,6 +14,7 @@
 class TestGalaxyInvocations(GalaxyTestBase.GalaxyTestBase):
     workflow_id: str
     pause_workflow_id: str
+    x_random_lines_workflow_id: str
 
     @classmethod
     def setUpClass(cls) -> None:
@@ -22,6 +23,8 @@ def setUpClass(cls) -> None:
         cls.workflow_id = cls.gi.workflows.import_workflow_from_local_path(path)["id"]
         path = test_util.get_abspath(os.path.join("data", "test_workflow_pause.ga"))
         cls.pause_workflow_id = cls.gi.workflows.import_workflow_from_local_path(path)["id"]
+        path = test_util.get_abspath(os.path.join("data", "select_x_random_lines.ga"))
+        cls.x_random_lines_workflow_id = cls.gi.workflows.import_workflow_from_local_path(path)["id"]
 
     def setUp(self):
         super().setUp()
@@ -172,6 +175,27 @@ def test_rerun_invocation(self):
         history = self.gi.histories.show_history(rerun_invocation["history_id"], contents=True)
         assert len(history) == 3
 
+    @test_util.skip_unless_galaxy("release_21.01")
+    def test_rerun_invocation_with_input_params(self):
+        threeline_dataset_id = self._test_dataset(self.history_id, contents="A\nB\nC")
+        invocation = self._invoke_x_random_lines_workflow(threeline_dataset_id)
+        self.gi.invocations.wait_for_invocation(invocation["id"])
+        rerun_invocation = self.gi.invocations.rerun_invocation(invocation["id"], history_id=self.history_id)
+        self.gi.invocations.wait_for_invocation(rerun_invocation["id"])
+
+    @test_util.skip_unless_galaxy("release_24.2")
+    def test_rerun_invocation_with_input_params_changed(self):
+        threeline_dataset_id = self._test_dataset(self.history_id, contents="A\nB\nC")
+        invocation = self._invoke_x_random_lines_workflow(threeline_dataset_id)
+        self.gi.invocations.wait_for_invocation(invocation["id"])
+        inputs_update = {"how_many": 1}
+        rerun_invocation = self.gi.invocations.rerun_invocation(
+            invocation["id"], inputs_update=inputs_update, history_id=self.history_id
+        )
+        self.gi.invocations.wait_for_invocation(rerun_invocation["id"])
+        rerun_request = self.gi.invocations.get_invocation_request(rerun_invocation["id"])
+        assert rerun_request["inputs"]["how_many"] == 1
+
     def _invoke_workflow(self) -> dict[str, Any]:
         dataset = {"src": "hda", "id": self.dataset_id}
 
@@ -188,3 +212,11 @@ def _invoke_pause_workflow(self) -> dict[str, Any]:
             inputs={"0": {"src": "hda", "id": self.dataset_id}},
             history_id=self.history_id,
         )
+
+    def _invoke_x_random_lines_workflow(self, dataset_id: str) -> dict[str, Any]:
+        return self.gi.workflows.invoke_workflow(
+            self.x_random_lines_workflow_id,
+            inputs={"from_what": {"src": "hda", "id": dataset_id}, "how_many": 2},
+            history_id=self.history_id,
+            inputs_by="name",
+        )

bioblend/_tests/data/select_x_random_lines.ga

@@ -0,0 +1,107 @@
+{
+    "a_galaxy_workflow": "true",
+    "annotation": "",
+    "comments": [],
+    "format-version": "0.1",
+    "name": "select X random lines",
+    "report": {
+        "markdown": "\n# Workflow Execution Report\n\n## Workflow Inputs\n```galaxy\ninvocation_inputs()\n```\n\n## Workflow Outputs\n```galaxy\ninvocation_outputs()\n```\n\n## Workflow\n```galaxy\nworkflow_display()\n```\n"
+    },
+    "steps": {
+        "0": {
+            "annotation": "",
+            "content_id": null,
+            "errors": null,
+            "id": 0,
+            "input_connections": {},
+            "inputs": [
+                {
+                    "description": "",
+                    "name": "how_many"
+                }
+            ],
+            "label": "how_many",
+            "name": "Input parameter",
+            "outputs": [],
+            "position": {
+                "left": 0.5625,
+                "top": 0
+            },
+            "tool_id": null,
+            "tool_state": "{\"validators\": [{\"min\": null, \"max\": null, \"negate\": false, \"type\": \"in_range\"}], \"parameter_type\": \"integer\", \"optional\": false}",
+            "tool_version": null,
+            "type": "parameter_input",
+            "uuid": "b609b67d-5101-4162-9307-51c0b7c48b51",
+            "when": null,
+            "workflow_outputs": []
+        },
+        "1": {
+            "annotation": "",
+            "content_id": null,
+            "errors": null,
+            "id": 1,
+            "input_connections": {},
+            "inputs": [
+                {
+                    "description": "",
+                    "name": "from_what"
+                }
+            ],
+            "label": "from_what",
+            "name": "Input dataset",
+            "outputs": [],
+            "position": {
+                "left": 0,
+                "top": 100.85546875
+            },
+            "tool_id": null,
+            "tool_state": "{\"optional\": false, \"tag\": null}",
+            "tool_version": null,
+            "type": "data_input",
+            "uuid": "0c3e99d4-8da2-4188-8aa9-cf9982561659",
+            "when": null,
+            "workflow_outputs": []
+        },
+        "2": {
+            "annotation": "",
+            "content_id": "random_lines1",
+            "errors": null,
+            "id": 2,
+            "input_connections": {
+                "input": {
+                    "id": 1,
+                    "output_name": "output"
+                },
+                "num_lines": {
+                    "id": 0,
+                    "output_name": "output"
+                }
+            },
+            "inputs": [],
+            "label": "random_lines",
+            "name": "Select random lines",
+            "outputs": [
+                {
+                    "name": "out_file1",
+                    "type": "input"
+                }
+            ],
+            "position": {
+                "left": 302.984375,
+                "top": 19.515625
+            },
+            "post_job_actions": {},
+            "tool_id": "random_lines1",
+            "tool_state": "{\"input\": {\"__class__\": \"ConnectedValue\"}, \"num_lines\": {\"__class__\": \"ConnectedValue\"}, \"seed_source\": {\"seed_source_selector\": \"no_seed\", \"__current_case__\": 0}, \"__page__\": 0, \"__rerun_remap_job_id__\": null}",
+            "tool_uuid": null,
+            "tool_version": "2.0.2",
+            "type": "tool",
+            "uuid": "6182ab34-7b61-49b1-9827-e1b1f81c57cc",
+            "when": null,
+            "workflow_outputs": []
+        }
+    },
+    "tags": [],
+    "uuid": "6b2bd605-08e4-467f-a854-e4681e7b62d0",
+    "version": 2
+}

bioblend/galaxy/invocations/__init__.py

@@ -172,6 +172,8 @@ def rerun_invocation(
         allow_tool_state_corrections: bool = False,
         inputs_by: Optional[InputsBy] = None,
         parameters_normalized: bool = False,
+        resource_params: Optional[dict[str, Any]] = None,
+        use_cached_job: bool = True,
     ) -> dict[str, Any]:
         """
         Rerun a workflow invocation. For more extensive documentation of all
@@ -181,9 +183,10 @@ def rerun_invocation(
         :param invocation_id: Encoded workflow invocation ID to be rerun
 
         :type inputs_update: dict
-        :param inputs_update: If different datasets should be used to the original
+        :param inputs_update: If different inputs should be used to the original
           invocation, this should contain a mapping of workflow inputs to the new
-          datasets and dataset collections.
+          datasets and dataset collections. Watch out for conflict with the
+          legacy params_update.
 
         :type params_update: dict
         :param params_update: If different non-dataset tool parameters should be
@@ -226,24 +229,53 @@ def rerun_invocation(
           Default is ``False``, but when setting parameters for a subworkflow,
           ``True`` is required.
 
+        :type resource_params: dict
+        :param resource_params: A dictionary containing the resource parameters
+          to be used for this workflow run.
+
+        :type use_cached_job: bool
+        :param use_cached_job: Whether to use cached jobs for the workflow
+          invocation.
+
         :rtype: dict
         :return: A dict describing the new workflow invocation.
 
         .. note::
           This method works only on Galaxy 21.01 or later.
         """
-        invocation_details = self.show_invocation(invocation_id)
-        workflow_id = invocation_details["workflow_id"]
-        inputs = invocation_details["inputs"]
-        wf_params = invocation_details["input_step_parameters"]
+        try:
+            payload = self.get_invocation_request(invocation_id)
+        except ConnectionError as e:
+            if e.status_code != 404:
+                raise
+            # Galaxy release_24.1 or earlier
+            invocation = self.show_invocation(invocation_id)
+            workflow_step_id_to_index = {
+                step["workflow_step_id"]: index for index, step in enumerate(invocation["steps"])
+            }
+            # Merge input_step_parameters (indexed by label) into inputs (indexed by step index)
+            inputs = invocation["inputs"]
+            for param_input_dict in invocation["input_step_parameters"].values():
+                workflow_step_id = param_input_dict["workflow_step_id"]
+                workflow_step_index = workflow_step_id_to_index[workflow_step_id]
+                inputs[str(workflow_step_index)] = param_input_dict
+            payload = {
+                "inputs": inputs,
+                "instance": True,
+                "workflow_id": invocation["workflow_id"],
+            }
+        else:
+            # Drop history_id from the payload as we will set history later
+            payload.pop("history_id")
+        workflow_id = payload["workflow_id"]
         if inputs_update:
-            for inp, input_value in inputs_update.items():
-                inputs[inp] = input_value
+            if payload.get("inputs") is None:
+                payload["inputs"] = {}
+            payload["inputs"].update(inputs_update)
         if params_update:
-            for param, param_value in params_update.items():
-                wf_params[param] = param_value
-        payload = {"inputs": inputs, "params": wf_params}
-
+            if payload.get("parameters") is None:
+                payload["parameters"] = {}
+            payload["parameters"].update(params_update)
         if replacement_params:
             payload["replacement_params"] = replacement_params
         if history_id:
@@ -258,9 +290,11 @@ def rerun_invocation(
             payload["inputs_by"] = inputs_by
         if parameters_normalized:
             payload["parameters_normalized"] = parameters_normalized
-        api_params = {"instance": True}
+        if resource_params:
+            payload["resource_params"] = resource_params
+        payload["use_cached_job"] = use_cached_job
         url = "/".join((self.gi.url, "workflows", workflow_id, "invocations"))
-        return self.gi.make_post_request(url=url, payload=payload, params=api_params)
+        return self.gi.make_post_request(url=url, payload=payload)
 
     def cancel_invocation(self, invocation_id: str) -> dict[str, Any]:
         """
@@ -374,6 +408,22 @@ def get_invocation_step_jobs_summary(self, invocation_id: str) -> list[dict[str,
         url = self._make_url(invocation_id) + "/step_jobs_summary"
         return self._get(url=url)
 
+    def get_invocation_request(self, invocation_id: str) -> dict[str, Any]:
+        """
+        Get a request dict for an invocation.
+
+        :type invocation_id: str
+        :param invocation_id: Encoded workflow invocation ID
+
+        :rtype: dict
+        :return: The invocation request.
+
+        .. note::
+          This method works only on Galaxy 24.2 or later.
+        """
+        url = self._make_url(invocation_id) + "/request"
+        return self._get(url=url)
+
     def get_invocation_report(self, invocation_id: str) -> dict[str, Any]:
         """
         Get a Markdown report for an invocation.