Merge pull request #496 from mvdbeek/gxformat2_export

Add gxformat2 export

CHANGELOG.md

@@ -5,6 +5,9 @@
 * Added ``offset`` parameter to ``InvocationClient.get_invocations()`` and
   BioBlend.objects ``ObjInvocationClient.list()`` methods.
 
+* Added ``style`` parameter to ``WorkflowClient.export_workflow_dict()`` method
+  (thanks to [Marius van den Beek](https://github.com/mvdbeek)).
+
 * Improvements to type annotations and documentation.
 
 ### BioBlend v1.4.0 - 2024-11-06

bioblend/_tests/TestGalaxyWorkflows.py

@@ -3,7 +3,11 @@
 import shutil
 import tempfile
 import time
-from typing import Any
+from typing import (
+    Any,
+    Literal,
+    Optional,
+)
 
 import pytest
 
@@ -141,7 +145,7 @@ def test_import_publish_workflow_from_local_path(self):
         assert not imported_wf["deleted"]
         assert imported_wf["published"]
 
-    def test_import_export_workflow_dict(self):
+    def _import_export(self, style: Optional[Literal["ga", "format2"]] = None):
         path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
         with open(path) as f:
             wf_dict = json.load(f)
@@ -151,8 +155,21 @@ def test_import_export_workflow_dict(self):
         assert imported_wf["url"].startswith("/api/workflows/")
         assert not imported_wf["deleted"]
         assert not imported_wf["published"]
-        exported_wf_dict = self.gi.workflows.export_workflow_dict(imported_wf["id"])
+        exported_wf_dict = self.gi.workflows.export_workflow_dict(imported_wf["id"], style=style)
         assert isinstance(exported_wf_dict, dict)
+        if style == "format2":
+            assert exported_wf_dict["class"] == "GalaxyWorkflow"
+        else:
+            assert exported_wf_dict["a_galaxy_workflow"] == "true"
+
+    def test_import_export_workflow_dict(self):
+        self._import_export()
+
+    def test_import_export_workflow_dict_ga(self):
+        self._import_export("ga")
+
+    def test_import_export_workflow_dict_format2(self):
+        self._import_export("format2")
 
     def test_import_publish_workflow_dict(self):
         path = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))

bioblend/galaxy/workflows/__init__.py

@@ -11,6 +11,8 @@
     TYPE_CHECKING,
 )
 
+import yaml
+
 from bioblend.galaxy.client import Client
 
 if TYPE_CHECKING:
@@ -193,7 +195,9 @@ def import_shared_workflow(self, workflow_id: str) -> dict[str, Any]:
         url = self._make_url()
         return self._post(url=url, payload=payload)
 
-    def export_workflow_dict(self, workflow_id: str, version: Optional[int] = None) -> dict[str, Any]:
+    def export_workflow_dict(
+        self, workflow_id: str, version: Optional[int] = None, style: Optional[Literal["ga", "format2"]] = None
+    ) -> dict[str, Any]:
         """
         Exports a workflow.
 
@@ -203,15 +207,24 @@ def export_workflow_dict(self, workflow_id: str, version: Optional[int] = None)
         :type version: int
         :param version: Workflow version to export
 
+        :type style
+        :param style: Either "ga" for the original JSON format or "format2" for
+          the modern YAML gxformat2 format.
+
         :rtype: dict
         :return: Dictionary representing the requested workflow
         """
         params: dict[str, Any] = {}
         if version is not None:
             params["version"] = version
-
+        if style:
+            params["style"] = style
         url = "/".join((self._make_url(), "download", workflow_id))
-        return self._get(url=url, params=params)
+        json = style != "format2"
+        response = self._get(url=url, params=params, json=json)
+        if not json:
+            return yaml.safe_load(response.text)
+        return response
 
     def export_workflow_to_local_path(
         self, workflow_id: str, file_local_path: str, use_default_filename: bool = True

setup.cfg

@@ -68,6 +68,7 @@ disable_error_code = func-returns-value
 
 [options]
 install_requires =
+    PyYAML
     requests>=2.20.0
     requests-toolbelt>=0.5.1,!=0.9.0
     tuspy

tox.ini

@@ -43,4 +43,5 @@ deps =
     mypy
     ruff
     types-requests
+    types-PyYAML
 skip_install = true