Use structural pattern matching

Author: mr-c

cwltool/argparser.py

@@ -3,7 +3,7 @@
 import argparse
 import os
 import urllib
-from collections.abc import Callable, MutableMapping, MutableSequence, Sequence
+from collections.abc import Callable, MutableSequence, Sequence
 from typing import Any, cast
 
 import rich.markup
@@ -897,55 +897,51 @@ def add_argument(
     atype: Any | None = None
     typekw: dict[str, Any] = {}
 
-    if inptype == "File":
-        action = FileAction
-    elif inptype == "Directory":
-        action = DirectoryAction
-    elif isinstance(inptype, MutableMapping) and inptype["type"] == "array":
-        if inptype["items"] == "File":
+    match inptype:
+        case "File":
+            action = FileAction
+        case "Directory":
+            action = DirectoryAction
+        case {"type": "array", "items": "File"}:
             action = FileAppendAction
-        elif inptype["items"] == "Directory":
+        case {"type": "array", "items": "Directory"}:
             action = DirectoryAppendAction
-        else:
+        case {"type": "array", "items": str(items)}:
             action = AppendAction
-            items = inptype["items"]
-            if items == "int" or items == "long":
-                atype = int
-            elif items == "double" or items == "float":
-                atype = float
-    elif isinstance(inptype, MutableMapping) and inptype["type"] == "enum":
-        atype = str
-    elif isinstance(inptype, MutableMapping) and inptype["type"] == "record":
-        records.append(name)
-        for field in inptype["fields"]:
-            fieldname = name + "." + shortname(field["name"])
-            fieldtype = field["type"]
-            fielddescription = field.get("doc", "")
-            add_argument(
-                toolparser,
-                fieldname,
-                fieldtype,
-                records,
-                fielddescription,
-                default=default.get(shortname(field["name"]), None) if default else None,
-                input_required=required,
-            )
-        return
-    elif inptype == "string":
-        atype = str
-    elif inptype == "int":
-        atype = int
-    elif inptype == "long":
-        atype = int
-    elif inptype == "double":
-        atype = float
-    elif inptype == "float":
-        atype = float
-    elif inptype == "boolean":
-        action = "store_true"
-    else:
-        _logger.debug("Can't make command line argument from %s", inptype)
-        return None
+            match items:
+                case "int" | "long":
+                    atype = int
+                case "double" | "float":
+                    atype = float
+        case {"type": "enum"}:
+            atype = str
+        case {"type": "record", "fields": list(fields)}:
+            records.append(name)
+            for field in fields:
+                fieldname = name + "." + shortname(field["name"])
+                fieldtype = field["type"]
+                fielddescription = field.get("doc", "")
+                add_argument(
+                    toolparser,
+                    fieldname,
+                    fieldtype,
+                    records,
+                    fielddescription,
+                    default=default.get(shortname(field["name"]), None) if default else None,
+                    input_required=required,
+                )
+            return
+        case "string":
+            atype = str
+        case "int" | "long":
+            atype = int
+        case "double" | "float":
+            atype = float
+        case "boolean":
+            action = "store_true"
+        case _:
+            _logger.debug("Can't make command line argument from %s", inptype)
+            return None
 
     if action in (FileAction, DirectoryAction, FileAppendAction, DirectoryAppendAction):
         typekw["urljoin"] = urljoin

cwltool/builder.py

@@ -166,6 +166,10 @@ def build_job_script(self, commands: list[str]) -> str | None:
             return self.job_script_provider.build_job_script(self, commands)
         return None
 
+    def _capture_files(self, f: CWLObjectType) -> CWLObjectType:
+        self.files.append(f)
+        return f
+
     def bind_input(
         self,
         schema: CWLObjectType,
@@ -555,7 +559,7 @@ def addsf(
                 visit_class(
                     datum.get("secondaryFiles", []),
                     ("File", "Directory"),
-                    _capture_files,
+                    self._capture_files,
                 )
 
             if schema["type"] == "org.w3id.cwl.cwl.Directory":
@@ -570,7 +574,7 @@ def addsf(
                 self.files.append(datum)
 
             if schema["type"] == "Any":
-                visit_class(datum, ("File", "Directory"), _capture_files)
+                visit_class(datum, ("File", "Directory"), self._capture_files)
 
         # Position to front of the sort key
         if binding:
@@ -589,23 +593,21 @@ def tostr(self, value: MutableMapping[str, str] | Any) -> str:
         :raises WorkflowException: if the item is a File or Directory and the
           "path" is missing.
         """
-        if isinstance(value, MutableMapping) and value.get("class") in (
-            "File",
-            "Directory",
-        ):
-            if "path" not in value:
-                raise WorkflowException(
-                    '{} object missing "path": {}'.format(value["class"], value)
-                )
-            return value["path"]
-        elif isinstance(value, ScalarFloat):
-            rep = RoundTripRepresenter()
-            dec_value = Decimal(rep.represent_scalar_float(value).value)
-            if "E" in str(dec_value):
-                return str(dec_value.quantize(1))
-            return str(dec_value)
-        else:
-            return str(value)
+        match value:
+            case {"class": "File" | "Directory" as class_name, **rest}:
+                if "path" not in rest:
+                    raise WorkflowException(
+                        '{} object missing "path": {}'.format(class_name, value)
+                    )
+                return str(rest["path"])
+            case ScalarFloat():
+                rep = RoundTripRepresenter()
+                dec_value = Decimal(rep.represent_scalar_float(value).value)
+                if "E" in str(dec_value):
+                    return str(dec_value.quantize(1))
+                return str(dec_value)
+            case _:
+                return str(value)
 
     def generate_arg(self, binding: CWLObjectType) -> list[str]:
         """Convert an input binding to a list of command line arguments."""
@@ -632,30 +634,28 @@ def generate_arg(self, binding: CWLObjectType) -> list[str]:
                 raise WorkflowException("'separate' option can not be specified without prefix")
 
         argl: MutableSequence[CWLOutputType] = []
-        if isinstance(value, MutableSequence):
-            if binding.get("itemSeparator") and value:
-                itemSeparator = cast(str, binding["itemSeparator"])
-                argl = [itemSeparator.join([self.tostr(v) for v in value])]
-            elif binding.get("valueFrom"):
-                value = [self.tostr(v) for v in value]
-                return cast(list[str], ([prefix] if prefix else [])) + cast(list[str], value)
-            elif prefix and value:
+        match value:
+            case MutableSequence():
+                if binding.get("itemSeparator") and value:
+                    itemSeparator = cast(str, binding["itemSeparator"])
+                    argl = [itemSeparator.join([self.tostr(v) for v in value])]
+                elif binding.get("valueFrom"):
+                    value = [self.tostr(v) for v in value]
+                    return cast(list[str], ([prefix] if prefix else [])) + cast(list[str], value)
+                elif prefix and value:
+                    return [prefix]
+                else:
+                    return []
+            case {"class": "File" | "Directory"}:
+                argl = [cast(CWLOutputType, value)]
+            case MutableMapping():
+                return [prefix] if prefix else []
+            case True if prefix:
                 return [prefix]
-            else:
+            case bool() | None:
                 return []
-        elif isinstance(value, MutableMapping) and value.get("class") in (
-            "File",
-            "Directory",
-        ):
-            argl = cast(MutableSequence[CWLOutputType], [value])
-        elif isinstance(value, MutableMapping):
-            return [prefix] if prefix else []
-        elif value is True and prefix:
-            return [prefix]
-        elif value is False or value is None or (value is True and not prefix):
-            return []
-        else:
-            argl = [value]
+            case _:
+                argl = [value]
 
         args = []
         for j in argl:

cwltool/checker.py

@@ -33,31 +33,37 @@ def check_types(
     """
     if valueFrom is not None:
         return "pass"
-    if linkMerge is None:
-        if can_assign_src_to_sink(srctype, sinktype, strict=True):
-            return "pass"
-        if can_assign_src_to_sink(srctype, sinktype, strict=False):
-            return "warning"
-        return "exception"
-    if linkMerge == "merge_nested":
-        return check_types(
-            {"items": _get_type(srctype), "type": "array"},
-            _get_type(sinktype),
-            None,
-            None,
-        )
-    if linkMerge == "merge_flattened":
-        return check_types(merge_flatten_type(_get_type(srctype)), _get_type(sinktype), None, None)
-    raise WorkflowException(f"Unrecognized linkMerge enum {linkMerge!r}")
+    match linkMerge:
+        case None:
+            if can_assign_src_to_sink(srctype, sinktype, strict=True):
+                return "pass"
+            if can_assign_src_to_sink(srctype, sinktype, strict=False):
+                return "warning"
+            return "exception"
+        case "merge_nested":
+            return check_types(
+                {"items": _get_type(srctype), "type": "array"},
+                _get_type(sinktype),
+                None,
+                None,
+            )
+        case "merge_flattened":
+            return check_types(
+                merge_flatten_type(_get_type(srctype)), _get_type(sinktype), None, None
+            )
+        case _:
+            raise WorkflowException(f"Unrecognized linkMerge enum {linkMerge!r}")
 
 
 def merge_flatten_type(src: SinkType) -> CWLOutputType:
     """Return the merge flattened type of the source type."""
-    if isinstance(src, MutableSequence):
-        return [merge_flatten_type(t) for t in src]
-    if isinstance(src, MutableMapping) and src.get("type") == "array":
-        return src
-    return {"items": src, "type": "array"}
+    match src:
+        case MutableSequence():
+            return [merge_flatten_type(t) for t in src]
+        case {"type": "array"}:
+            return src
+        case _:
+            return {"items": src, "type": "array"}
 
 
 def can_assign_src_to_sink(src: SinkType, sink: SinkType | None, strict: bool = False) -> bool:

cwltool/command_line_tool.py

@@ -1388,13 +1388,14 @@ def collect_output(
 
             optional = False
             single = False
-            if isinstance(schema["type"], MutableSequence):
-                if "null" in schema["type"]:
-                    optional = True
-                if "File" in schema["type"] or "Directory" in schema["type"]:
+            match schema["type"]:
+                case MutableSequence():
+                    if "null" in schema["type"]:
+                        optional = True
+                    if "File" in schema["type"] or "Directory" in schema["type"]:
+                        single = True
+                case "File" | "Directory":
                     single = True
-            elif schema["type"] == "File" or schema["type"] == "Directory":
-                single = True
 
             if "outputEval" in binding:
                 with SourceLine(binding, "outputEval", WorkflowException, debug):