Add gxformat2 export

Author: mvdbeek

bioblend/_tests/TestGalaxyWorkflows.py

@@ -7,6 +7,8 @@
     Any,
     Dict,
     List,
+    Literal,
+    Optional,
 )
 
 import pytest
@@ -145,7 +147,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)
@@ -155,8 +157,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

@@ -29,7 +29,11 @@ def __init__(self, galaxy_instance: "GalaxyInstance") -> None:
 
     # the 'deleted' option is not available for workflows
     def get_workflows(
-        self, workflow_id: Optional[str] = None, name: Optional[str] = None, published: bool = False
+        self,
+        workflow_id: Optional[str] = None,
+        name: Optional[str] = None,
+        published: bool = False,
+        style: Optional[Literal["ga", "format2"]] = None,
     ) -> List[Dict[str, Any]]:
         """
         Get all workflows, or select a subset by specifying optional arguments
@@ -41,6 +45,9 @@ def get_workflows(
         :type published: bool
         :param published: if ``True``, return also published workflows
 
+        :type style
+        :param style: Set style to `format2` to get gxformat2 encoded workflow
+
         :rtype: list
         :return: A list of workflow dicts.
                  For example::
@@ -60,6 +67,8 @@ def get_workflows(
         params: Dict[str, Any] = {}
         if published:
             params["show_published"] = True
+        if style:
+            params["style"] = style
         workflows = self._get(params=params)
         if name is not None:
             workflows = [_ for _ in workflows if _["name"] == name]