formatting

Author: mr-c

cwl_utils/graph_split.py

@@ -10,26 +10,25 @@
 
 import argparse
 import json
+import logging
 import os
+import re
 import sys
 from io import TextIOWrapper
 from pathlib import Path
 from typing import (
     IO,
     TYPE_CHECKING,
     Any,
-    List,
     MutableMapping,
+    Optional,
     Set,
+    TextIO,
     Union,
     cast,
-    Optional, TextIO,
 )
-import logging
-import re
 
 from cwlformat.formatter import stringify_dict
-from ruamel.yaml.comments import Format
 from ruamel.yaml.main import YAML
 from ruamel.yaml.representer import RoundTripRepresenter
 from schema_salad.sourceline import SourceLine, add_lc_filename
@@ -103,11 +102,11 @@ def run(args: list[str]) -> int:
 
 
 def graph_split(
-        sourceIO: IO[str],
-        output_dir: "StrPath",
-        output_format: str,
-        mainfile: str,
-        pretty: bool,
+    sourceIO: IO[str],
+    output_dir: "StrPath",
+    output_format: str,
+    mainfile: str,
+    pretty: bool,
 ) -> None:
     """Loop over the provided packed CWL document and split it up."""
     yaml = YAML(typ="rt")
@@ -129,7 +128,7 @@ def graph_split(
         os.mkdir(output_dir)
 
     def my_represent_none(
-            self: Any, data: Any
+        self: Any, data: Any
     ) -> Any:  # pylint: disable=unused-argument
         """Force clean representation of 'null'."""
         return self.represent_scalar("tag:yaml.org,2002:null", "null")
@@ -156,7 +155,9 @@ def my_represent_none(
             yaml_dump(entry, output_file, pretty)
 
 
-def rewrite(document: Any, doc_id: str, output_dir: Path, pretty: Optional[bool] = False) -> Set[str]:
+def rewrite(
+    document: Any, doc_id: str, output_dir: Path, pretty: Optional[bool] = False
+) -> Set[str]:
     """Rewrite the given element from the CWL $graph."""
     imports = set()
     if isinstance(document, list) and not isinstance(document, str):
@@ -169,27 +170,27 @@ def rewrite(document: Any, doc_id: str, output_dir: Path, pretty: Optional[bool]
                 if key == "run" and isinstance(value, str) and value[0] == "#":
                     document[key] = f"{re.sub('.cwl$', '', value[1:])}.cwl"
                 elif key in ("id", "outputSource") and value.startswith("#" + doc_id):
-                    document[key] = value[len(doc_id) + 2:]
+                    document[key] = value[len(doc_id) + 2 :]
                 elif key == "out" and isinstance(value, list):
 
                     def rewrite_id(entry: Any) -> Union[MutableMapping[Any, Any], str]:
                         if isinstance(entry, MutableMapping):
                             if entry["id"].startswith(this_id):
                                 assert isinstance(this_id, str)  # nosec B101
-                                entry["id"] = cast(str, entry["id"])[len(this_id) + 1:]
+                                entry["id"] = cast(str, entry["id"])[len(this_id) + 1 :]
                             return entry
                         elif isinstance(entry, str):
                             if this_id and entry.startswith(this_id):
-                                return entry[len(this_id) + 1:]
+                                return entry[len(this_id) + 1 :]
                             return entry
                         raise Exception(f"{entry} is neither a dictionary nor string.")
 
                     document[key][:] = [rewrite_id(entry) for entry in value]
                 elif key in ("source", "scatter", "items", "format"):
                     if (
-                            isinstance(value, str)
-                            and value.startswith("#")
-                            and "/" in value
+                        isinstance(value, str)
+                        and value.startswith("#")
+                        and "/" in value
                     ):
                         referrant_file, sub = value[1:].split("/", 1)
                         if referrant_file == doc_id:
@@ -200,7 +201,7 @@ def rewrite_id(entry: Any) -> Union[MutableMapping[Any, Any], str]:
                         new_sources = list()
                         for entry in value:
                             if entry.startswith("#" + doc_id):
-                                new_sources.append(entry[len(doc_id) + 2:])
+                                new_sources.append(entry[len(doc_id) + 2 :])
                             else:
                                 new_sources.append(entry)
                         document[key] = new_sources
@@ -231,7 +232,7 @@ def rewrite_types(field: Any, entry_file: str, sameself: bool) -> None:
                 if key == name:
                     if isinstance(value, str) and value.startswith(entry_file):
                         if sameself:
-                            field[key] = value[len(entry_file) + 1:]
+                            field[key] = value[len(entry_file) + 1 :]
                         else:
                             field[key] = "{d[0]}#{d[1]}".format(
                                 d=value[1:].split("/", 1)
@@ -243,7 +244,9 @@ def rewrite_types(field: Any, entry_file: str, sameself: bool) -> None:
                     rewrite_types(entry, entry_file, sameself)
 
 
-def rewrite_schemadef(document: MutableMapping[str, Any], output_dir: Path, pretty: Optional[bool] = False) -> Set[str]:
+def rewrite_schemadef(
+    document: MutableMapping[str, Any], output_dir: Path, pretty: Optional[bool] = False
+) -> Set[str]:
     """Dump the schemadefs to their own file."""
     for entry in document["types"]:
         if "$import" in entry:
@@ -281,7 +284,11 @@ def json_dump(entry: Any, output_file: str) -> None:
         json.dump(entry, result_handle, indent=4)
 
 
-def yaml_dump(entry: Any, output_file_or_handle: Optional[Union[str, Path, TextIOWrapper, TextIO]], pretty: bool) -> None:
+def yaml_dump(
+    entry: Any,
+    output_file_or_handle: Optional[Union[str, Path, TextIOWrapper, TextIO]],
+    pretty: bool,
+) -> None:
     """Output object as YAML."""
     yaml = YAML(typ="rt", pure=True)
     yaml.default_flow_style = False
@@ -293,21 +300,16 @@ def yaml_dump(entry: Any, output_file_or_handle: Optional[Union[str, Path, TextI
             if pretty:
                 result_handle.write(stringify_dict(entry))
             else:
-                yaml.dump(
-                    entry,
-                    result_handle
-                )
+                yaml.dump(entry, result_handle)
     elif isinstance(output_file_or_handle, (TextIOWrapper, TextIO)):
         if pretty:
             output_file_or_handle.write(stringify_dict(entry))
         else:
-            yaml.dump(
-                entry,
-                output_file_or_handle
-            )
+            yaml.dump(entry, output_file_or_handle)
     else:
         raise ValueError(
-            f"output_file_or_handle must be a string or a file handle but got {type(output_file_or_handle)}")
+            f"output_file_or_handle must be a string or a file handle but got {type(output_file_or_handle)}"
+        )
 
 
 if __name__ == "__main__":