gxformat2 abstraction layer...

lib/galaxy/managers/workflows.py

@@ -14,13 +14,6 @@
 )
 
 import sqlalchemy
-import yaml
-from gxformat2 import (
-    from_galaxy_native,
-    ImporterGalaxyInterface,
-    ImportOptions,
-    python_to_workflow,
-)
 from gxformat2.abstract import from_dict
 from gxformat2.cytoscape import to_cytoscape
 from gxformat2.yaml import ordered_dump
@@ -113,6 +106,10 @@
     RawTextTerm,
 )
 from galaxy.work.context import WorkRequestContext
+from galaxy.workflow.format2 import (
+    convert_from_format2,
+    convert_to_format2,
+)
 from galaxy.workflow.modules import (
     module_factory,
     ToolModule,
@@ -603,10 +600,7 @@ def read_workflow_from_path(self, app, user, path, allow_in_directory=None) -> m
         workflow_class, as_dict, object_id = artifact_class(trans, as_dict, allow_in_directory=allow_in_directory)
         assert workflow_class == "GalaxyWorkflow"
         # Format 2 Galaxy workflow.
-        galaxy_interface = Format2ConverterGalaxyInterface()
-        import_options = ImportOptions()
-        import_options.deduplicate_subworkflows = True
-        as_dict = python_to_workflow(as_dict, galaxy_interface, workflow_directory=None, import_options=import_options)
+        as_dict = convert_from_format2(as_dict, None)
         raw_description = RawWorkflowDescription(as_dict, path)
         created_workflow = self.build_workflow_from_raw_description(trans, raw_description, WorkflowCreateOptions())
         return created_workflow.workflow
@@ -633,15 +627,7 @@ def normalize_workflow_format(self, trans, as_dict):
         workflow_class, as_dict, object_id = artifact_class(trans, as_dict)
         if workflow_class == "GalaxyWorkflow" or "yaml_content" in as_dict:
             # Format 2 Galaxy workflow.
-            galaxy_interface = Format2ConverterGalaxyInterface()
-            import_options = ImportOptions()
-            import_options.deduplicate_subworkflows = True
-            try:
-                as_dict = python_to_workflow(
-                    as_dict, galaxy_interface, workflow_directory=workflow_directory, import_options=import_options
-                )
-            except yaml.scanner.ScannerError as e:
-                raise exceptions.MalformedContents(str(e))
+            as_dict = convert_from_format2(as_dict, workflow_directory)
 
         return RawWorkflowDescription(as_dict, workflow_path)
 
@@ -906,8 +892,8 @@ def workflow_to_dict(self, trans, stored, style="export", version=None, history=
         fields like 'url' and 'url' and actual unencoded step ids instead of 'order_index'.
         """
 
-        def to_format_2(wf_dict, **kwds):
-            return from_galaxy_native(wf_dict, None, **kwds)
+        def to_format_2(wf_dict, json_wrapper: bool):
+            return convert_to_format2(wf_dict, json_wrapper=json_wrapper)
 
         if version == "":
             version = None
@@ -928,7 +914,7 @@ def to_format_2(wf_dict, **kwds):
             wf_dict = self._workflow_to_dict_preview(trans, workflow=workflow)
         elif style == "format2":
             wf_dict = self._workflow_to_dict_export(trans, stored, workflow=workflow)
-            wf_dict = to_format_2(wf_dict)
+            wf_dict = to_format_2(wf_dict, json_wrapper=False)
         elif style == "format2_wrapped_yaml":
             wf_dict = self._workflow_to_dict_export(trans, stored, workflow=workflow)
             wf_dict = to_format_2(wf_dict, json_wrapper=True)
@@ -978,7 +964,7 @@ def store_workflow_to_path(self, workflow_path, stored_workflow, workflow, **kwd
                 ordered_dump(abstract_dict, f)
             else:
                 wf_dict = self._workflow_to_dict_export(trans, stored_workflow, workflow=workflow)
-                wf_dict = from_galaxy_native(wf_dict, None, json_wrapper=True)
+                wf_dict = convert_to_format2(wf_dict, json_wrapper=True)
                 f.write(wf_dict["yaml_content"])
 
     def _workflow_to_dict_run(self, trans, stored, workflow, history=None):
@@ -2099,13 +2085,6 @@ def __init__(self, as_dict, workflow_path=None):
         self.workflow_path = workflow_path
 
 
-class Format2ConverterGalaxyInterface(ImporterGalaxyInterface):
-    def import_workflow(self, workflow, **kwds):
-        raise NotImplementedError(
-            "Direct format 2 import of nested workflows is not yet implemented, use bioblend client."
-        )
-
-
 def _get_stored_workflow(session, workflow_uuid, workflow_id, by_stored_id):
     stmt = select(StoredWorkflow)
     if workflow_uuid is not None:

lib/galaxy/workflow/format2.py

@@ -0,0 +1,36 @@
+from typing import Optional
+
+import yaml
+from gxformat2 import (
+    from_galaxy_native,
+    ImporterGalaxyInterface,
+    ImportOptions,
+    python_to_workflow,
+)
+
+from galaxy.exceptions import MalformedContents
+
+
+def convert_to_format2(as_dict, json_wrapper: bool):
+    return from_galaxy_native(as_dict, None, json_wrapper=json_wrapper)
+
+
+def convert_from_format2(as_dict, workflow_directory: Optional[str]):
+    # Format 2 Galaxy workflow.
+    galaxy_interface = Format2ConverterGalaxyInterface()
+    import_options = ImportOptions()
+    import_options.deduplicate_subworkflows = True
+    try:
+        as_dict = python_to_workflow(
+            as_dict, galaxy_interface, workflow_directory=workflow_directory, import_options=import_options
+        )
+    except yaml.scanner.ScannerError as e:
+        raise MalformedContents(str(e))
+    return as_dict
+
+
+class Format2ConverterGalaxyInterface(ImporterGalaxyInterface):
+    def import_workflow(self, workflow, **kwds):
+        raise NotImplementedError(
+            "Direct format 2 import of nested workflows is not yet implemented, use bioblend client."
+        )

test/unit/workflows/test_convert.py

@@ -0,0 +1,35 @@
+from galaxy.workflow.format2 import (
+    convert_from_format2,
+    convert_to_format2,
+)
+
+WORKFLOW_WITH_OUTPUTS = """
+class: GalaxyWorkflow
+inputs:
+  input1: data
+outputs:
+  wf_output_1:
+    outputSource: first_cat/out_file1
+steps:
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input1
+      queries_0|input2: input1
+"""
+
+
+def test_convert_from_simple():
+    as_dict = {"yaml_content": WORKFLOW_WITH_OUTPUTS}
+    ga_format = convert_from_format2(as_dict, None)
+    assert ga_format["a_galaxy_workflow"] == "true"
+    assert ga_format["format-version"] == "0.1"
+
+
+def test_convert_to_simple():
+    as_dict = {"yaml_content": WORKFLOW_WITH_OUTPUTS}
+    ga_format = convert_from_format2(as_dict, None)
+
+    as_dict = convert_to_format2(ga_format, True)
+    assert "yaml_content" in as_dict
+    assert "yaml_content" not in as_dict