black 20.8b1

Author: mr-c

cwltool/builder.py

@@ -519,7 +519,9 @@ def addsf(
                 ll = schema.get("loadListing") or self.loadListing
                 if ll and ll != "no_listing":
                     get_listing(
-                        self.fs_access, datum, (ll == "deep_listing"),
+                        self.fs_access,
+                        datum,
+                        (ll == "deep_listing"),
                     )
                 self.files.append(datum)
 

cwltool/command_line_tool.py

@@ -485,7 +485,8 @@ def _initialworkdir(self, j: JobBase, builder: Builder) -> None:
             # "listing" is an array of either expressions or Dirent so
             # evaluate each item
             for t in cast(
-                MutableSequence[Union[str, CWLObjectType]], initialWorkdir["listing"],
+                MutableSequence[Union[str, CWLObjectType]],
+                initialWorkdir["listing"],
             ):
                 if isinstance(t, Mapping) and "entry" in t:
                     # Dirent
@@ -672,7 +673,9 @@ def remove_dirname(d: CWLObjectType) -> None:
                             del d["dirname"]
 
                     visit_class(
-                        entry["listing"], ("File", "Directory"), remove_dirname,
+                        entry["listing"],
+                        ("File", "Directory"),
+                        remove_dirname,
                     )
 
             visit_class(

cwltool/cwlrdf.py

@@ -195,7 +195,5 @@ def printdot(
     ctx: ContextType,
     stdout: Union[TextIO, StreamWriter],
 ) -> None:
-    cwl_viewer = CWLViewer(printrdf(wf, ctx, 'n3'))  # type: CWLViewer
-    stdout.write(
-        cwl_viewer.dot()
-    )
+    cwl_viewer = CWLViewer(printrdf(wf, ctx, "n3"))  # type: CWLViewer
+    stdout.write(cwl_viewer.dot())

cwltool/cwlviewer.py

@@ -8,119 +8,149 @@
 class CWLViewer:
     """Produce similar images with the https://github.com/common-workflow-language/cwlviewer."""
 
-    _queries_dir = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'rdfqueries')
-    _get_inner_edges_query_path = os.path.join(_queries_dir, 'get_inner_edges.sparql')
-    _get_input_edges_query_path = os.path.join(_queries_dir, 'get_input_edges.sparql')
-    _get_output_edges_query_path = os.path.join(_queries_dir, 'get_output_edges.sparql')
-    _get_root_query_path = os.path.join(_queries_dir, 'get_root.sparql')
+    _queries_dir = os.path.join(
+        os.path.abspath(os.path.dirname(__file__)), "rdfqueries"
+    )
+    _get_inner_edges_query_path = os.path.join(_queries_dir, "get_inner_edges.sparql")
+    _get_input_edges_query_path = os.path.join(_queries_dir, "get_input_edges.sparql")
+    _get_output_edges_query_path = os.path.join(_queries_dir, "get_output_edges.sparql")
+    _get_root_query_path = os.path.join(_queries_dir, "get_root.sparql")
 
     def __init__(
-            self,
-            rdf_description  # type: str
+        self, rdf_description  # type: str
     ):
         """Create a viewer object based on the rdf description of the workflow."""
         self._dot_graph = CWLViewer._init_dot_graph()  # type: pydot.Graph
-        self._rdf_graph = self._load_cwl_graph(rdf_description)  # type: rdflib.graph.Graph
+        self._rdf_graph = self._load_cwl_graph(
+            rdf_description
+        )  # type: rdflib.graph.Graph
         self._root_graph_uri = self._get_root_graph_uri()  # type: str
         self._set_inner_edges()
         self._set_input_edges()
         self._set_output_edges()
 
     def _load_cwl_graph(
-            self,
-            rdf_description  # type: str
+        self, rdf_description  # type: str
     ) -> rdflib.graph.Graph:
         rdf_graph = rdflib.Graph()
-        rdf_graph.parse(data=rdf_description, format='n3')
+        rdf_graph.parse(data=rdf_description, format="n3")
         return rdf_graph
 
     def _set_inner_edges(self) -> None:
         with open(self._get_inner_edges_query_path) as f:
             get_inner_edges_query = f.read()
-        inner_edges = self._rdf_graph.query(get_inner_edges_query, initBindings={'root_graph': self._root_graph_uri})
+        inner_edges = self._rdf_graph.query(
+            get_inner_edges_query, initBindings={"root_graph": self._root_graph_uri}
+        )
         for inner_edge_row in inner_edges:
-            source_label = inner_edge_row['source_label'] \
-                if inner_edge_row['source_label'] is not None \
-                else urlparse(inner_edge_row['source_step']).fragment
+            source_label = (
+                inner_edge_row["source_label"]
+                if inner_edge_row["source_label"] is not None
+                else urlparse(inner_edge_row["source_step"]).fragment
+            )
             n = pydot.Node(
-                '',
-                fillcolor='lightgoldenrodyellow', style="filled",
+                "",
+                fillcolor="lightgoldenrodyellow",
+                style="filled",
                 label=source_label,
-                shape='record'
+                shape="record",
             )
-            n.set_name(str(inner_edge_row['source_step']))
+            n.set_name(str(inner_edge_row["source_step"]))
             self._dot_graph.add_node(n)
-            target_label = inner_edge_row['target_label'] \
-                if inner_edge_row['target_label'] is not None \
-                else urlparse(inner_edge_row['target_step']).fragment
+            target_label = (
+                inner_edge_row["target_label"]
+                if inner_edge_row["target_label"] is not None
+                else urlparse(inner_edge_row["target_step"]).fragment
+            )
             n = pydot.Node(
-                '',
-                fillcolor='lightgoldenrodyellow', style="filled",
+                "",
+                fillcolor="lightgoldenrodyellow",
+                style="filled",
                 label=target_label,
-                shape='record'
+                shape="record",
             )
-            n.set_name(str(inner_edge_row['target_step']))
+            n.set_name(str(inner_edge_row["target_step"]))
             self._dot_graph.add_node(n)
-            self._dot_graph.add_edge(pydot.Edge(str(inner_edge_row['source_step']), str(inner_edge_row['target_step'])))
+            self._dot_graph.add_edge(
+                pydot.Edge(
+                    str(inner_edge_row["source_step"]),
+                    str(inner_edge_row["target_step"]),
+                )
+            )
 
     def _set_input_edges(self) -> None:
         with open(self._get_input_edges_query_path) as f:
             get_input_edges_query = f.read()
         inputs_subgraph = pydot.Subgraph(graph_name="cluster_inputs")
         self._dot_graph.add_subgraph(inputs_subgraph)
         inputs_subgraph.set_rank("same")
-        inputs_subgraph.create_attribute_methods(['style'])
+        inputs_subgraph.create_attribute_methods(["style"])
         inputs_subgraph.set_style("dashed")
         inputs_subgraph.set_label("Workflow Inputs")
 
-        input_edges = self._rdf_graph.query(get_input_edges_query, initBindings={'root_graph': self._root_graph_uri})
+        input_edges = self._rdf_graph.query(
+            get_input_edges_query, initBindings={"root_graph": self._root_graph_uri}
+        )
         for input_row in input_edges:
             n = pydot.Node(
-                '',
+                "",
                 fillcolor="#94DDF4",
                 style="filled",
-                label=urlparse(input_row['input']).fragment,
-                shape='record'
+                label=urlparse(input_row["input"]).fragment,
+                shape="record",
             )
-            n.set_name(str(input_row['input']))
+            n.set_name(str(input_row["input"]))
             inputs_subgraph.add_node(n)
-            self._dot_graph.add_edge(pydot.Edge(str(input_row['input']), str(input_row['step'])))
+            self._dot_graph.add_edge(
+                pydot.Edge(str(input_row["input"]), str(input_row["step"]))
+            )
 
     def _set_output_edges(self) -> None:
         with open(self._get_output_edges_query_path) as f:
             get_output_edges = f.read()
         outputs_graph = pydot.Subgraph(graph_name="cluster_outputs")
         self._dot_graph.add_subgraph(outputs_graph)
         outputs_graph.set_rank("same")
-        outputs_graph.create_attribute_methods(['style'])
+        outputs_graph.create_attribute_methods(["style"])
         outputs_graph.set_style("dashed")
         outputs_graph.set_label("Workflow Outputs")
         outputs_graph.set_labelloc("b")
-        output_edges = self._rdf_graph.query(get_output_edges, initBindings={'root_graph': self._root_graph_uri})
+        output_edges = self._rdf_graph.query(
+            get_output_edges, initBindings={"root_graph": self._root_graph_uri}
+        )
         for output_edge_row in output_edges:
-            n = pydot.Node('',
-                           fillcolor="#94DDF4",
-                           style="filled",
-                           label=urlparse(output_edge_row['output']).fragment,
-                           shape='record'
-                           )
-            n.set_name(str(output_edge_row['output']))
+            n = pydot.Node(
+                "",
+                fillcolor="#94DDF4",
+                style="filled",
+                label=urlparse(output_edge_row["output"]).fragment,
+                shape="record",
+            )
+            n.set_name(str(output_edge_row["output"]))
             outputs_graph.add_node(n)
-            self._dot_graph.add_edge(pydot.Edge(output_edge_row['step'], output_edge_row['output']))
+            self._dot_graph.add_edge(
+                pydot.Edge(output_edge_row["step"], output_edge_row["output"])
+            )
 
     def _get_root_graph_uri(self) -> rdflib.URIRef:
         with open(self._get_root_query_path) as f:
             get_root_query = f.read()
-        root = list(self._rdf_graph.query(get_root_query, ))
+        root = list(
+            self._rdf_graph.query(
+                get_root_query,
+            )
+        )
         if len(root) != 1:
-            raise RuntimeError("Cannot identify root workflow! Notice that only Workflows can be visualized")
+            raise RuntimeError(
+                "Cannot identify root workflow! Notice that only Workflows can be visualized"
+            )
 
-        workflow = root[0]['workflow']  # type: rdflib.URIRef
+        workflow = root[0]["workflow"]  # type: rdflib.URIRef
         return workflow
 
     @classmethod
     def _init_dot_graph(cls) -> pydot.Graph:
-        graph = pydot.Graph(graph_type='digraph', simplify=False)
+        graph = pydot.Graph(graph_type="digraph", simplify=False)
         graph.set_bgcolor("#eeeeee")
         graph.set_clusterrank("local")
         graph.set_labelloc("bottom")

cwltool/executors.py

@@ -339,7 +339,9 @@ def _runner(self, job, runtime_context, TMPDIR_LOCK):
                     runtime_context.workflow_eval_lock.notifyAll()
 
     def run_job(
-        self, job: Optional[JobsType], runtime_context: RuntimeContext,
+        self,
+        job: Optional[JobsType],
+        runtime_context: RuntimeContext,
     ) -> None:
         """Execute a single Job in a seperate thread."""
         if job is not None:
@@ -368,11 +370,10 @@ def run_job(
                         return
 
                     if (
-                        (self.allocated_ram + job.builder.resources["ram"])
-                        > self.max_ram
-                        or (self.allocated_cores + job.builder.resources["cores"])
-                        > self.max_cores
-                    ):
+                        self.allocated_ram + job.builder.resources["ram"]
+                    ) > self.max_ram or (
+                        self.allocated_cores + job.builder.resources["cores"]
+                    ) > self.max_cores:
                         _logger.debug(
                             'Job "%s" cannot run yet, resources (%s) are not '
                             "available (already allocated ram is %d, allocated cores is %d, "

cwltool/job.py

@@ -608,7 +608,9 @@ def get_from_requirements(
 
     @abstractmethod
     def create_runtime(
-        self, env: MutableMapping[str, str], runtime_context: RuntimeContext,
+        self,
+        env: MutableMapping[str, str],
+        runtime_context: RuntimeContext,
     ) -> Tuple[List[str], Optional[str]]:
         """Return the list of commands to run the selected container engine."""
 

cwltool/load_tool.py

@@ -482,7 +482,9 @@ def load_tool(
 
 
 def resolve_overrides(
-    ov: IdxResultType, ov_uri: str, baseurl: str,
+    ov: IdxResultType,
+    ov_uri: str,
+    baseurl: str,
 ) -> List[CWLObjectType]:
     ovloader = Loader(overrides_ctx)
     ret, _ = ovloader.resolve_all(ov, baseurl)

cwltool/main.py

@@ -132,7 +132,8 @@ def _signal_handler(signum: int, _: Any) -> None:
 
 
 def generate_example_input(
-    inptype: Optional[CWLOutputType], default: Optional[CWLOutputType],
+    inptype: Optional[CWLOutputType],
+    default: Optional[CWLOutputType],
 ) -> Tuple[Any, str]:
     """Convert a single input schema into an example."""
     example = None
@@ -401,7 +402,8 @@ def init_job_order(
         if "job_order" in cmd_line and cmd_line["job_order"]:
             try:
                 job_order_object = cast(
-                    CWLObjectType, loader.resolve_ref(cmd_line["job_order"])[0],
+                    CWLObjectType,
+                    loader.resolve_ref(cmd_line["job_order"])[0],
                 )
             except Exception:
                 _logger.exception(
@@ -582,7 +584,10 @@ def loadref(base: str, uri: str) -> Union[CommentedMap, CommentedSeq, str, None]
     return deps
 
 
-def print_pack(loadingContext: LoadingContext, uri: str,) -> str:
+def print_pack(
+    loadingContext: LoadingContext,
+    uri: str,
+) -> str:
     """Return a CWL serialization of the CWL document in JSON."""
     packed = pack(loadingContext, uri)
     if len(cast(Sized, packed["$graph"])) > 1:
@@ -666,7 +671,9 @@ def formatTime(
 
 
 def setup_provenance(
-    args: argparse.Namespace, argsl: List[str], runtimeContext: RuntimeContext,
+    args: argparse.Namespace,
+    argsl: List[str],
+    runtimeContext: RuntimeContext,
 ) -> Optional[int]:
     if not args.compute_checksum:
         _logger.error("--provenance incompatible with --no-compute-checksum")
@@ -719,7 +726,9 @@ def setup_loadingContext(
     return loadingContext
 
 
-def make_template(tool: Process,) -> None:
+def make_template(
+    tool: Process,
+) -> None:
     def my_represent_none(
         self: Any, data: Any
     ) -> Any:  # pylint: disable=unused-argument
@@ -737,7 +746,9 @@ def my_represent_none(
 
 
 def choose_target(
-    args: argparse.Namespace, tool: Process, loadingContext: LoadingContext,
+    args: argparse.Namespace,
+    tool: Process,
+    loadingContext: LoadingContext,
 ) -> Optional[Process]:
 
     if loadingContext.loader is None:
@@ -769,7 +780,9 @@ def choose_target(
     return tool
 
 
-def check_working_directories(runtimeContext: RuntimeContext,) -> Optional[int]:
+def check_working_directories(
+    runtimeContext: RuntimeContext,
+) -> Optional[int]:
     for dirprefix in ("tmpdir_prefix", "tmp_outdir_prefix", "cachedir"):
         if (
             getattr(runtimeContext, dirprefix)

cwltool/pack.py

@@ -29,7 +29,9 @@
 
 
 def find_run(
-    d: Union[CWLObjectType, ResolveType], loadref: LoadRefType, runs: Set[str],
+    d: Union[CWLObjectType, ResolveType],
+    loadref: LoadRefType,
+    runs: Set[str],
 ) -> None:
     if isinstance(d, MutableSequence):
         for s in d:

cwltool/pathmapper.py

@@ -118,7 +118,10 @@ def visit(
     ) -> None:
         stagedir = cast(Optional[str], obj.get("dirname")) or stagedir
         tgt = convert_pathsep_to_unix(
-            os.path.join(stagedir, cast(str, obj["basename"]),)
+            os.path.join(
+                stagedir,
+                cast(str, obj["basename"]),
+            )
         )
         if obj["location"] in self._pathmap:
             return
@@ -212,7 +215,10 @@ def files(self) -> List[str]:
     def items(self) -> List[Tuple[str, MapperEnt]]:
         return list(self._pathmap.items())
 
-    def reversemap(self, target: str,) -> Optional[Tuple[str, str]]:
+    def reversemap(
+        self,
+        target: str,
+    ) -> Optional[Tuple[str, str]]:
         for k, v in self._pathmap.items():
             if v[1] == target:
                 return (k, v[0])