function used to import an invocation archive. Needs API updates before it can be merged.

Author: Smeds

bioblend/_tests/TestGalaxyInvocations.py

@@ -112,6 +112,15 @@ def test_get_invocation_jobs_summary(self):
         assert len(step_jobs_summary) == 1
         assert step_jobs_summary[0]["populated_state"] == "ok"
 
+    @test_util.skip_unless_galaxy("release_23.0")
+    def test_import_invocation_archive(self):
+        new_history = self.gi.histories.create_history(name="My History")
+        test_file = test_util.get_abspath(os.path.join("data", "example3.rocrate.zip"))
+        import_response = self.gi.invocations.import_invocation(
+            history_id=new_history["id"], model_store_format="rocrate.zip", file_path=test_file
+        )
+        assert "model_class" in import_response[0] and import_response[0]["model_class"] == "WorkflowInvocation"
+
     @test_util.skip_unless_galaxy("release_19.09")
     @test_util.skip_unless_tool("cat1")
     @test_util.skip_unless_tool("cat")

bioblend/_tests/data/example3.rocrate.zip

@@ -2,6 +2,7 @@
 Contains possible interactions with the Galaxy workflow invocations
 """
 
+import base64
 import logging
 from typing import (
     Any,
@@ -267,6 +268,39 @@ def cancel_invocation(self, invocation_id: str) -> dict[str, Any]:
         url = self._make_url(invocation_id)
         return self._delete(url=url)
 
+    def import_invocation(
+        self, history_id: str, model_store_format: str, file_path: Optional[str] = None, url: Optional[str] = None
+    ) -> Any:
+        """
+        Import a invocation from an archive on disk or a URL.
+
+        :type history_id
+        :param history_id: id of the history where the invocation will be imported
+
+        :type model_store_format
+        :param model_store_format: archive type that will be imported
+
+        :type file_path: str
+        :param file_path: Path to exported history archive on disk.
+
+        :type url: str
+        :param url: URL for an exported history archive
+
+        :rtype: dict or list of dicts
+        :return: if the import is successful, a list of dictionaries will be returned; otherwise, a single dictionary will be returned.
+        """
+        payload: dict[str, Any] = {
+            "history_id": history_id,
+            "model_store_format": model_store_format,
+        }
+        if file_path:
+            with open(file_path, "rb") as reader:
+                payload["store_content_uri"] = "base64://" + base64.b64encode(reader.read()).decode("utf-8")
+        else:
+            payload["store_content_uri"] = url
+        url = "/".join((self._make_url(), "from_store"))
+        return self._post(url=url, payload=payload)
+
     def show_invocation_step(self, invocation_id: str, step_id: str) -> dict[str, Any]:
         """
         See the details of a particular workflow invocation step.