Minor readability refactor (PolusAI#342)

Author: vjaganat90

src/sophios/api/pythonapi.py

@@ -405,34 +405,39 @@ def _validate(self) -> None:
     def _yml(self) -> dict:
         in_dict: dict[str, Any] = {}  # NOTE: input values can be arbitrary JSON; not just strings!
         for inp in self.inputs:
-            if inp.value is not None:
-                if isinstance(inp.value, Path):
-                    # Special case for Path since it does not inherit from YAMLObject
-                    in_dict[inp.name] = str(inp.value)
-                elif isinstance(inp.value, dict) and isinstance(inp.value.get('wic_alias', {}), Path):
-                    # Special case for Path since it does not inherit from YAMLObject
-                    in_dict[inp.name] = {'wic_alias': str(inp.value['wic_alias'])}
-                elif isinstance(inp.value, dict) and isinstance(inp.value.get('wic_inline_input', {}), Path):
-                    # Special case for Path since it does not inherit from YAMLObject
-                    in_dict[inp.name] = {'wic_inline_input': str(inp.value['wic_inline_input'])}
-                elif isinstance(inp.value, dict) and isinstance(inp.value.get('wic_inline_input', {}), str):
-                    # Special case for inline str since it does not inherit from YAMLObject
-                    in_dict[inp.name] = {'wic_inline_input': inp.value.get('wic_inline_input')}
-                elif isinstance(inp.value, str):
-                    in_dict[inp.name] = inp.value  # Obviously strings are serializable
-                elif isinstance(inp.value, yaml.YAMLObject):
-                    # Serialization and deserialization logic should always be
-                    # encapsulated within each object. For the pyyaml library,
-                    # each object should inherit from pyyaml.YAMLObject.
-                    # See https://pyyaml.org/wiki/PyYAMLDocumentation
-                    # Section "Constructors, representers, resolvers"
-                    # class Monster(yaml.YAMLObject): ...
-                    in_dict[inp.name] = inp.value
-                else:
-                    logger.warning(f'Warning! input name {inp.name} input value {inp.value}')
-                    logger.warning('is not an instance of YAMLObject. The default str() serialization')
-                    logger.warning('logic often gives bad results. Please explicitly inherit from YAMLObject.')
-                    in_dict[inp.name] = inp.value
+            if inp.value:
+                match inp.value:
+                    case Path():
+                        # Special case for Path since it does not inherit from YAMLObject
+                        in_dict[inp.name] = str(inp.value)
+                    case {'wic_alias': wic_alias, **rest_of_dict}:
+                        match wic_alias:
+                            case Path():
+                                # Special case for Path since it does not inherit from YAMLObject
+                                in_dict[inp.name] = {'wic_alias': str(inp.value['wic_alias'])}
+                    case {'wic_inline_input': wic_inline_input, **rest_of_dict}:
+                        match wic_inline_input:
+                            case Path():
+                                # Special case for Path since it does not inherit from YAMLObject
+                                in_dict[inp.name] = {'wic_inline_input': str(inp.value['wic_inline_input'])}
+                            case str():
+                                # Special case for inline str since it does not inherit from YAMLObject
+                                in_dict[inp.name] = {'wic_inline_input': inp.value.get('wic_inline_input')}
+                    case str():
+                        in_dict[inp.name] = inp.value  # Obviously strings are serializable
+                    case yaml.YAMLObject():
+                        # Serialization and deserialization logic should always be
+                        # encapsulated within each object. For the pyyaml library,
+                        # each object should inherit from pyyaml.YAMLObject.
+                        # See https://pyyaml.org/wiki/PyYAMLDocumentation
+                        # Section "Constructors, representers, resolvers"
+                        # class Monster(yaml.YAMLObject): ...
+                        in_dict[inp.name] = inp.value
+                    case _:
+                        logger.warning(f'Warning! input name {inp.name} input value {inp.value}')
+                        logger.warning('is not an instance of YAMLObject. The default str() serialization')
+                        logger.warning('logic often gives bad results. Please explicitly inherit from YAMLObject.')
+                        in_dict[inp.name] = inp.value
 
         out_list: list = []  # The out: tag is a list, not a dict
         out_list = [{out.name: out.value} for out in self.outputs if out.value]
@@ -523,11 +528,13 @@ def _validate(self) -> None:
         # subworkflows will NOT have all required inputs.
         for s in self.steps:
             try:
-                if isinstance(s, Step):
-                    s._validate()  # pylint: disable=W0212
-                if isinstance(s, Workflow):
-                    # recursively validate subworkflows ?
-                    s._validate()  # pylint: disable=W0212
+                match s:
+                    case Step():
+                        s._validate()
+                    case Workflow():
+                        s._validate()
+                    case _:
+                        pass
             except BaseException as exc:
                 raise InvalidStepError(
                     f"{s.process_name} is missing required inputs"
@@ -558,21 +565,23 @@ def yaml(self) -> dict[str, Any]:
         # TODO: outputs?
         steps = []
         for s in self.steps:
-            if isinstance(s, Step):
-                steps.append(s._yml)
-            elif isinstance(s, Workflow):
-                ins = {
-                    inp.name: inp.value
-                    for inp in s.inputs
-                    if inp.value is not None  # Subworkflow args are not required
-                }
-                parentargs: dict[str, Any] = {"in": ins} if ins else {}
-                # See the second to last line of ast.read_ast_from_disk()
-                d = {'id': self.process_name + '.wic',
-                     'subtree': s.yaml,  # recursively call .yaml (i.e. on s, not self)
-                     'parentargs': parentargs}
-                steps.append(d)
-            #  else: ...
+            match s:
+                case Step():
+                    steps.append(s._yml)
+                case Workflow() as sw:
+                    ins = {
+                        inp.name: inp.value
+                        for inp in sw.inputs
+                        if inp.value is not None  # Subworkflow args are not required
+                    }
+                    parentargs: dict[str, Any] = {"in": ins} if ins else {}
+                    # See the second to last line of ast.read_ast_from_disk()
+                    d = {'id': self.process_name + '.wic',
+                         'subtree': sw.yaml,  # recursively call .yaml (i.e. on s, not self)
+                         'parentargs': parentargs}
+                    steps.append(d)
+                case _:
+                    pass
         yaml_contents = {"inputs": inputs, "steps": steps} if inputs else {"steps": steps}
         return yaml_contents
 
@@ -700,10 +709,11 @@ def flatten_steps(self) -> list[Step]:
         """
         steps = []
         for step in self.steps:
-            if isinstance(step, Step):
-                steps.append(step)
-            if isinstance(step, Workflow):
-                steps += step.flatten_steps()
+            match step:
+                case Step():
+                    steps.append(step)
+                case Workflow():
+                    steps += step.flatten_steps()
         return steps
 
     # NOTE: Cannot return list[Workflow] because Workflow is not yet defined.
@@ -773,8 +783,6 @@ def get_cwl_workflow(self, args_dict: Dict[str, str] = {}) -> Json:
         Returns:
             Json: Contains the compiled CWL and yaml inputs to the workflow.
         """
-        user_args = self._convert_args_dict_to_args_list(args_dict)
-        args = get_args(self.process_name, user_args)
         compiler_info = self.compile(args_dict=args_dict, write_to_disk=False)
         rose_tree = compiler_info.rose
 

src/sophios/api/utils/ict/ict_spec/tools/cwl_ict.py

@@ -53,12 +53,11 @@ def clt_dict(ict_: "ICT", network_access: bool) -> dict:
 
 def remove_none(d: Union[dict, str]) -> Union[dict, str]:
     """Recursively remove keys with None values."""
-    if isinstance(d, dict):
-        return {k: remove_none(v) for k, v in d.items() if v is not None}
-    elif isinstance(d, str):
-        return d  # Return the string unchanged
-    else:
-        return d  # Return other types of values unchanged
+    match d:
+        case dict():
+            return {k: remove_none(v) for k, v in d.items() if v is not None}
+        case str():
+            return d
 
 
 def input_output_dict(ict_: "ICT") -> Union[dict, str]:

src/sophios/main.py

@@ -39,8 +39,6 @@ def main() -> None:
                                       args.validate_plugins,
                                       not args.no_skip_dollar_schemas,
                                       args.quiet)
-    # This takes ~1 second but it is not really necessary.
-    # utils_graphs.make_plugins_dag(tools_cwl, args.graph_dark_theme)
     # pass around config object instead of reading from the disk!
     yml_paths = plugins.get_yml_paths(global_config)
 
@@ -55,7 +53,7 @@ def main() -> None:
 
     # Generating yml schemas every time takes ~20 seconds and guarantees the
     # subworkflow schemas are always up to date. However, since it compiles all
-    # yml files, if there are any errors in any of the yml files, the user may
+    # wic files, if there are any errors in any of the wic files, the user may
     # be confused by an error message when the --yaml file is correct.
     # For now, require the user to update the schemas manually. In the future,
     # we may use a filewatcher.