consolidate dict boilerplate into one function

Author: Vasu Jaganath

src/sophios/apis/python/api.py

@@ -15,7 +15,7 @@
 from sophios import compiler, input_output, plugins, utils_cwl
 from sophios import run_local as rl
 from sophios import post_compile as pc
-from sophios.cli import get_args, get_known_and_unknown_args
+from sophios.cli import get_args, get_known_and_unknown_args, get_dicts_for_compilation
 from sophios.utils_graphs import get_graph_reps
 from sophios.utils import convert_args_dict_to_args_list
 from sophios.wic_types import CompilerInfo, RoseTree, StepId, Tool, Tools, YamlTree, Json
@@ -720,28 +720,7 @@ def compile(self, write_to_disk: bool = False) -> CompilerInfo:
         steps_config = extract_tools_paths_NONPORTABLE(self.flatten_steps())
         global_config = merge(steps_config, global_config, strategy=Strategy.TYPESAFE_REPLACE)
 
-        # core compiler options for transformation into CWL
-        compiler_options: Dict[str, bool] = {}
-        compiler_options['partial_failure_enable'] = args.partial_failure_enable
-        compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-        compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-        compiler_options['inference_disable'] = args.inference_disable
-        compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-        # to be given to graph util functions
-        graph_settings: Dict[str, Any] = {}
-        graph_settings['graph_dark_theme'] = args.graph_dark_theme
-        graph_settings['graph_inline_depth'] = args.graph_inline_depth
-        graph_settings['graph_label_edges'] = args.graph_label_edges
-        graph_settings['graph_label_stepname'] = args.graph_label_stepname
-        graph_settings['graph_show_outputs'] = args.graph_show_outputs
-        graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-        # to be given to io absolute_yaml_tags function
-        yaml_tag_paths: Dict[str, str] = {}
-        yaml_tag_paths['cachedir'] = args.cachedir
-        yaml_tag_paths['yaml'] = args.yaml
-        yaml_tag_paths['homedir'] = args.homedir
+        compiler_options, graph_settings, yaml_tag_paths = get_dicts_for_compilation()
 
         # The compile_workflow function is 100% in-memory
         compiler_info = compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,

src/sophios/apis/rest/api.py

@@ -1,7 +1,6 @@
 from pathlib import Path
 import copy
 import yaml
-from typing import Dict, Any
 
 
 import uvicorn
@@ -14,7 +13,7 @@
 from sophios.utils_yaml import wic_loader
 from sophios import utils_cwl
 from sophios.post_compile import cwl_inline_runtag
-from sophios.cli import get_args
+from sophios.cli import get_args, get_dicts_for_compilation
 from sophios.wic_types import CompilerInfo, Json, Tool, Tools, StepId, YamlTree, Cwl, NodeData
 from sophios.apis.utils import converter
 import sophios.plugins as plugins
@@ -112,28 +111,7 @@ async def compile_wf(request: Request) -> Json:
     graph = get_graph_reps(wkflw_name)
     yaml_tree: YamlTree = YamlTree(StepId(wkflw_name, plugin_ns), workflow_can)
 
-    # core compiler options for transformation into CWL
-    compiler_options: Dict[str, bool] = {}
-    compiler_options['partial_failure_enable'] = args.partial_failure_enable
-    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-    compiler_options['inference_disable'] = args.inference_disable
-    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-    # to be given to graph util functions
-    graph_settings: Dict[str, Any] = {}
-    graph_settings['graph_dark_theme'] = args.graph_dark_theme
-    graph_settings['graph_inline_depth'] = args.graph_inline_depth
-    graph_settings['graph_label_edges'] = args.graph_label_edges
-    graph_settings['graph_label_stepname'] = args.graph_label_stepname
-    graph_settings['graph_show_outputs'] = args.graph_show_outputs
-    graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-    # to be given to io absolute_yaml_tags function
-    yaml_tag_paths: Dict[str, str] = {}
-    yaml_tag_paths['cachedir'] = args.cachedir
-    yaml_tag_paths['yaml'] = args.yaml
-    yaml_tag_paths['homedir'] = args.homedir
+    compiler_options, graph_settings, yaml_tag_paths = get_dicts_for_compilation()
 
     # ========= COMPILE WORKFLOW ================
     compiler_info: CompilerInfo = compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,

src/sophios/cli.py

@@ -1,7 +1,7 @@
 import argparse
 import sys
 from pathlib import Path
-from typing import List, Tuple
+from typing import List, Tuple, Dict, Any
 from unittest.mock import patch
 
 from . import _version
@@ -153,3 +153,36 @@ def get_known_and_unknown_args(yaml_path: str = '', suppliedargs: list[str] = []
     with patch.object(sys, 'argv', testargs):
         known_args, unknown_args = parser.parse_known_args()
     return known_args, unknown_args
+
+
+def get_dicts_for_compilation() -> Tuple[Dict[str, bool], Dict[str, Any], Dict[str, str]]:
+    """This is used to get default command line arguments for compilation
+    as a tuple of three dictionaries
+
+    Returns:
+        Tuple[Dict[str, bool], Dict[str,Any], Dict[str, str]]: The mocked command line arguments
+    """
+    args = get_args()
+    # core compiler options for transformation into CWL
+    compiler_options: Dict[str, bool] = {}
+    compiler_options['partial_failure_enable'] = args.partial_failure_enable
+    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
+    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
+    compiler_options['inference_disable'] = args.inference_disable
+    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
+
+    # to be given to graph util functions
+    graph_settings: Dict[str, Any] = {}
+    graph_settings['graph_dark_theme'] = args.graph_dark_theme
+    graph_settings['graph_inline_depth'] = args.graph_inline_depth
+    graph_settings['graph_label_edges'] = args.graph_label_edges
+    graph_settings['graph_label_stepname'] = args.graph_label_stepname
+    graph_settings['graph_show_outputs'] = args.graph_show_outputs
+    graph_settings['graph_show_inputs'] = args.graph_show_inputs
+
+    # to be given to io absolute_yaml_tags function
+    yaml_tag_paths: Dict[str, str] = {}
+    yaml_tag_paths['cachedir'] = args.cachedir
+    yaml_tag_paths['yaml'] = args.yaml
+    yaml_tag_paths['homedir'] = args.homedir
+    return (compiler_options, graph_settings, yaml_tag_paths)

src/sophios/cwl_subinterpreter.py

@@ -109,35 +109,15 @@ def rerun_cwltool(homedir: str, _directory_realtime: Path, cachedir_path: Path,
 
         # Setup dummy args
         args = cli.get_args()
+        compiler_options, graph_settings, yaml_tag_paths = cli.get_dicts_for_compilation()
 
         # TODO: Support other namespaces
         plugin_ns = 'global'  # wic['wic'].get('namespace', 'global')
         yaml_path = f'{cwl_tool}_only.wic'
         stepid = StepId(yaml_path, plugin_ns)
         yaml_tree = YamlTree(stepid, yml)
         subgraph = GraphReps(graphviz.Digraph(name=yaml_path), nx.DiGraph(), GraphData(yaml_path))
-        # core compiler options for transformation into CWL
-        compiler_options: Dict[str, bool] = {}
-        compiler_options['partial_failure_enable'] = args.partial_failure_enable
-        compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-        compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-        compiler_options['inference_disable'] = args.inference_disable
-        compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-        # to be given to graph util functions
-        graph_settings: Dict[str, Any] = {}
-        graph_settings['graph_dark_theme'] = args.graph_dark_theme
-        graph_settings['graph_inline_depth'] = args.graph_inline_depth
-        graph_settings['graph_label_edges'] = args.graph_label_edges
-        graph_settings['graph_label_stepname'] = args.graph_label_stepname
-        graph_settings['graph_show_outputs'] = args.graph_show_outputs
-        graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-        # to be given to io absolute_yaml_tags function
-        yaml_tag_paths: Dict[str, str] = {}
-        yaml_tag_paths['cachedir'] = args.cachedir
-        yaml_tag_paths['yaml'] = args.yaml
-        yaml_tag_paths['homedir'] = args.homedir
+
         compiler_info = compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,
                                                   [], [subgraph], {}, {}, {}, {},
                                                   tools_cwl, True, relative_run_path=False, testing=False)

src/sophios/main.py

@@ -143,28 +143,7 @@ def main() -> None:
         graphdata = GraphData(yaml_path)
         subgraph = GraphReps(subgraph_gv, subgraph_nx, graphdata)
 
-        # core compiler options for transformation into CWL
-        compiler_options: Dict[str, bool] = {}
-        compiler_options['partial_failure_enable'] = args.partial_failure_enable
-        compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-        compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-        compiler_options['inference_disable'] = args.inference_disable
-        compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-        # to be given to graph util functions
-        graph_settings: Dict[str, Any] = {}
-        graph_settings['graph_dark_theme'] = args.graph_dark_theme
-        graph_settings['graph_inline_depth'] = args.graph_inline_depth
-        graph_settings['graph_label_edges'] = args.graph_label_edges
-        graph_settings['graph_label_stepname'] = args.graph_label_stepname
-        graph_settings['graph_show_outputs'] = args.graph_show_outputs
-        graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-        # to be given to io absolute_yaml_tags function
-        yaml_tag_paths: Dict[str, str] = {}
-        yaml_tag_paths['cachedir'] = args.cachedir
-        yaml_tag_paths['yaml'] = args.yaml
-        yaml_tag_paths['homedir'] = args.homedir
+        compiler_options, graph_settings, yaml_tag_paths = cli.get_dicts_for_compilation()
 
         try:
             compiler_info = compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,

src/sophios/schemas/wic_schema.py

@@ -12,7 +12,7 @@
 
 import sophios
 from sophios import ast, compiler, utils_cwl
-from sophios.cli import get_args
+from sophios.cli import get_args, get_dicts_for_compilation
 from sophios.utils_yaml import wic_loader
 from sophios.wic_types import GraphData, GraphReps, NodeData, StepId, Yaml, YamlTree
 from ..wic_types import Json, Tools
@@ -640,28 +640,8 @@ def compile_workflow_generate_schema(homedir: str,
     graph = GraphReps(graph_gv, graph_nx, graphdata)
     args = get_args(str(yml_path), ['--allow_raw_cwl'] if allow_raw_cwl else [])
 
-    # core compiler options for transformation into CWL
-    compiler_options: Dict[str, bool] = {}
-    compiler_options['partial_failure_enable'] = args.partial_failure_enable
-    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-    compiler_options['inference_disable'] = args.inference_disable
-    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-    # to be given to graph util functions
-    graph_settings: Dict[str, Any] = {}
-    graph_settings['graph_dark_theme'] = args.graph_dark_theme
-    graph_settings['graph_inline_depth'] = args.graph_inline_depth
-    graph_settings['graph_label_edges'] = args.graph_label_edges
-    graph_settings['graph_label_stepname'] = args.graph_label_stepname
-    graph_settings['graph_show_outputs'] = args.graph_show_outputs
-    graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-    # to be given to io absolute_yaml_tags function
-    yaml_tag_paths: Dict[str, str] = {}
-    yaml_tag_paths['cachedir'] = args.cachedir
-    yaml_tag_paths['yaml'] = args.yaml
-    yaml_tag_paths['homedir'] = args.homedir
+    compiler_options, graph_settings, yaml_tag_paths = get_dicts_for_compilation()
+
     compiler_info = compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths, [], [graph],
                                               {}, {}, {}, {}, tools_cwl, True, relative_run_path=True, testing=True)
     rose_tree = compiler_info.rose

tests/test_examples.py

@@ -202,28 +202,7 @@ def run_workflows(yml_path_str: str, yml_path: Path, cwl_runner: str, args: argp
         with open(f'autogenerated/{Path(yml_path).stem}_tree_merged_inlined.wic', mode='w', encoding='utf-8') as f:
             f.write(yaml.dump(yaml_tree.yml))
 
-    # core compiler options for transformation into CWL
-    compiler_options: Dict[str, bool] = {}
-    compiler_options['partial_failure_enable'] = args.partial_failure_enable
-    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-    compiler_options['inference_disable'] = args.inference_disable
-    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-    # to be given to graph util functions
-    graph_settings = {}
-    graph_settings['graph_dark_theme'] = args.graph_dark_theme
-    graph_settings['graph_inline_depth'] = args.graph_inline_depth
-    graph_settings['graph_label_edges'] = args.graph_label_edges
-    graph_settings['graph_label_stepname'] = args.graph_label_stepname
-    graph_settings['graph_show_outputs'] = args.graph_show_outputs
-    graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-    # to be given to io absolute_yaml_tags function
-    yaml_tag_paths: Dict[str, str] = {}
-    yaml_tag_paths['cachedir'] = args.cachedir
-    yaml_tag_paths['yaml'] = args.yaml
-    yaml_tag_paths['homedir'] = args.homedir
+    compiler_options, graph_settings, yaml_tag_paths = sophios.cli.get_dicts_for_compilation()
 
     graph = get_graph_reps(str(yml_path))
     compiler_info = sophios.compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,
@@ -306,28 +285,8 @@ def test_cwl_embedding_independence(yml_path_str: str, yml_path: Path) -> None:
 
     graph = get_graph_reps(str(yml_path))
     is_root = True
-    # core compiler options for transformation into CWL
-    compiler_options: Dict[str, bool] = {}
-    compiler_options['partial_failure_enable'] = args.partial_failure_enable
-    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-    compiler_options['inference_disable'] = args.inference_disable
-    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-    # to be given to graph util functions
-    graph_settings = {}
-    graph_settings['graph_dark_theme'] = args.graph_dark_theme
-    graph_settings['graph_inline_depth'] = args.graph_inline_depth
-    graph_settings['graph_label_edges'] = args.graph_label_edges
-    graph_settings['graph_label_stepname'] = args.graph_label_stepname
-    graph_settings['graph_show_outputs'] = args.graph_show_outputs
-    graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-    # to be given to io absolute_yaml_tags function
-    yaml_tag_paths: Dict[str, str] = {}
-    yaml_tag_paths['cachedir'] = args.cachedir
-    yaml_tag_paths['yaml'] = args.yaml
-    yaml_tag_paths['homedir'] = args.homedir
+
+    compiler_options, graph_settings, yaml_tag_paths = sophios.cli.get_dicts_for_compilation()
 
     compiler_info = sophios.compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths,
                                                       [], [graph], {}, {}, {}, {}, tools_cwl,
@@ -434,28 +393,7 @@ def test_inline_subworkflows(yml_path_str: str, yml_path: Path) -> None:
     if namespaces_list == []:
         assert True  # There's nothing to test
 
-    # core compiler options for transformation into CWL
-    compiler_options: Dict[str, bool] = {}
-    compiler_options['partial_failure_enable'] = args.partial_failure_enable
-    compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-    compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-    compiler_options['inference_disable'] = args.inference_disable
-    compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
-
-    # to be given to graph util functions
-    graph_settings: Dict[str, Any] = {}
-    graph_settings['graph_dark_theme'] = args.graph_dark_theme
-    graph_settings['graph_inline_depth'] = args.graph_inline_depth
-    graph_settings['graph_label_edges'] = args.graph_label_edges
-    graph_settings['graph_label_stepname'] = args.graph_label_stepname
-    graph_settings['graph_show_outputs'] = args.graph_show_outputs
-    graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-    # to be given to io absolute_yaml_tags function
-    yaml_tag_paths: Dict[str, str] = {}
-    yaml_tag_paths['cachedir'] = args.cachedir
-    yaml_tag_paths['yaml'] = args.yaml
-    yaml_tag_paths['homedir'] = args.homedir
+    compiler_options, graph_settings, yaml_tag_paths = sophios.cli.get_dicts_for_compilation()
 
     graph = get_graph_reps(str(yml_path))
     compiler_info = sophios.compiler.compile_workflow(yaml_tree, compiler_options, graph_settings, yaml_tag_paths, [], [graph], {},

tests/test_fuzzy_compile.py

@@ -1,6 +1,5 @@
 from datetime import timedelta
 from pathlib import Path
-from typing import Dict, Any
 import unittest
 
 import graphviz
@@ -61,28 +60,8 @@ def test_fuzzy_compile(self, yml: Yaml) -> None:
         graph_nx = nx.DiGraph()
         graphdata = GraphData(str(yml_path))
         graph = GraphReps(graph_gv, graph_nx, graphdata)
-        # core compiler options for transformation into CWL
-        compiler_options: Dict[str, bool] = {}
-        compiler_options['partial_failure_enable'] = args.partial_failure_enable
-        compiler_options['inference_use_naming_conventions'] = args.inference_use_naming_conventions
-        compiler_options['insert_steps_automatically'] = args.insert_steps_automatically
-        compiler_options['inference_disable'] = args.inference_disable
-        compiler_options['allow_raw_cwl'] = args.allow_raw_cwl
 
-        # to be given to graph util functions
-        graph_settings: Dict[str, Any] = {}
-        graph_settings['graph_dark_theme'] = args.graph_dark_theme
-        graph_settings['graph_inline_depth'] = args.graph_inline_depth
-        graph_settings['graph_label_edges'] = args.graph_label_edges
-        graph_settings['graph_label_stepname'] = args.graph_label_stepname
-        graph_settings['graph_show_outputs'] = args.graph_show_outputs
-        graph_settings['graph_show_inputs'] = args.graph_show_inputs
-
-        # to be given to io absolute_yaml_tags function
-        yaml_tag_paths: Dict[str, str] = {}
-        yaml_tag_paths['cachedir'] = args.cachedir
-        yaml_tag_paths['yaml'] = args.yaml
-        yaml_tag_paths['homedir'] = args.homedir
+        compiler_options, graph_settings, yaml_tag_paths = sophios.cli.get_dicts_for_compilation()
 
         try:
             compiler_info = sophios.compiler.compile_workflow(yaml_tree, compiler_options, graph_settings,