function used to import an invocation archive.

bioblend/_tests/TestGalaxyInvocations.py

@@ -112,6 +112,18 @@ 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_25.01")
+    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,
+            upload_url="/upload/resumable_upload",
+        )
+        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/galaxy/invocations/__init__.py

@@ -267,6 +267,50 @@ 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,
+        upload_url: str = "/invocations/resumable_upload",
+        chunk_size: int = CHUNK_SIZE,
+    ) -> 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
+
+        :type chunk_size: int
+        :param chunk_size: Number of bytes to send in each chunk
+
+        :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:
+            uploader = self.gi.get_tus_uploader(path=file_path, url=upload_url, chunk_size=chunk_size)
+            uploader.upload()
+            assert uploader.session_id
+            payload["store_content_uri"] = f"tus://{uploader.session_id}"
+        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.