Blacken & isort

cylc/flow/pathutil.py

@@ -17,16 +17,25 @@
 
 import errno
 import os
-from pathlib import Path
 import re
+from pathlib import Path
 from shutil import rmtree
 from time import sleep
-from typing import Dict, Iterable, Set, Union, Optional, Any
+from typing import (
+    Any,
+    Dict,
+    Iterable,
+    Optional,
+    Set,
+    Union,
+)
 
 from cylc.flow import LOG
 from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
 from cylc.flow.exceptions import (
-    FileRemovalError, InputError, WorkflowFilesError
+    FileRemovalError,
+    InputError,
+    WorkflowFilesError,
 )
 from cylc.flow.platforms import get_localhost_install_target
 
@@ -40,7 +49,7 @@
     ^({re.escape(os.curdir)}|{re.escape(os.pardir)})
     ({re.escape(os.sep)}|$)
     ''',
-    re.VERBOSE
+    re.VERBOSE,
 )
 """Matches relative paths that are explicit (starts with ./)"""
 
@@ -58,9 +67,9 @@
 def expand_path(*args: Union[Path, str]) -> str:
     """Expand both vars and user in path and normalise it, joining any
     extra args."""
-    return os.path.normpath(os.path.expanduser(os.path.expandvars(
-        os.path.join(*args)
-    )))
+    return os.path.normpath(
+        os.path.expanduser(os.path.expandvars(os.path.join(*args)))
+    )
 
 
 def get_remote_workflow_run_dir(
@@ -87,7 +96,7 @@ def get_cylc_run_dir(alt_run_dir: Optional[str] = None) -> str:
 def get_alt_workflow_run_dir(
     alt_run_dir: Union[Path, str],
     workflow_id: Union[Path, str],
-    *args: Union[Path, str]
+    *args: Union[Path, str],
 ) -> str:
     """Return alternate workflow run directory.
 
@@ -125,9 +134,7 @@ def get_workflow_run_scheduler_log_path(workflow):
 
 def get_workflow_file_install_log_dir(workflow, *args):
     """Return workflow file install log file dir, join any extra args."""
-    return get_workflow_run_dir(
-        workflow, 'log', 'remote-install', *args
-    )
+    return get_workflow_run_dir(workflow, 'log', 'remote-install', *args)
 
 
 def get_workflow_run_config_log_dir(workflow, *args):
@@ -173,7 +180,7 @@ def make_workflow_run_tree(workflow):
 def make_localhost_symlinks(
     rund: Union[Path, str],
     named_sub_dir: str,
-    symlink_conf: Optional[Dict[str, Dict[str, str]]] = None
+    symlink_conf: Optional[Dict[str, Dict[str, str]]] = None,
 ) -> Dict[str, Union[Path, str]]:
     """Creates symlinks for any configured symlink dirs from glbl_cfg.
     Args:
@@ -189,7 +196,8 @@ def make_localhost_symlinks(
     symlinks_created = {}
     dirs_to_symlink = get_dirs_to_symlink(
         get_localhost_install_target(),
-        named_sub_dir, symlink_conf=symlink_conf
+        named_sub_dir,
+        symlink_conf=symlink_conf,
     )
     for key, value in dirs_to_symlink.items():
         if value is None:
@@ -204,7 +212,8 @@ def make_localhost_symlinks(
             raise WorkflowFilesError(
                 f"Can't symlink to {target}\n"
                 "Undefined variables, check "
-                f"global config: {', '.join(env_vars)}")
+                f"global config: {', '.join(env_vars)}"
+            )
 
         symlink_success = make_symlink_dir(symlink_path, target)
         # Symlink info returned for logging purposes. Symlinks should be
@@ -217,7 +226,7 @@ def make_localhost_symlinks(
 def get_dirs_to_symlink(
     install_target: str,
     workflow_id: str,
-    symlink_conf: Optional[Dict[str, Dict[str, Any]]] = None
+    symlink_conf: Optional[Dict[str, Dict[str, Any]]] = None,
 ) -> Dict[str, str]:
     """Returns dictionary of directories to symlink.
 
@@ -241,13 +250,15 @@ def get_dirs_to_symlink(
     base_dir = symlink_conf[install_target]['run']
     if base_dir:
         dirs_to_symlink['run'] = os.path.join(
-            base_dir, 'cylc-run', workflow_id)
+            base_dir, 'cylc-run', workflow_id
+        )
     for dir_ in SYMLINKABLE_LOCATIONS:
         link = symlink_conf[install_target].get(dir_, None)
         if (not link) or link == base_dir:
             continue
         dirs_to_symlink[dir_] = os.path.join(
-            link, 'cylc-run', workflow_id, dir_)
+            link, 'cylc-run', workflow_id, dir_
+        )
     return dirs_to_symlink
 
 
@@ -440,16 +451,17 @@ def get_next_rundir_number(run_path: Union[str, Path]) -> int:
         # This function unlikely to be called in circumstances where this will
         # be a problem.
         last_run_num = re_runX.search(  # type: ignore
-            old_run_path).group(1)
+            old_run_path
+        ).group(1)
         last_run_num = int(last_run_num)
     else:
         # If the ``runN`` symlink has been removed, get next numbered run from
         # file names:
         paths = Path(run_path).glob('run[0-9]*')
         run_numbers = (
-            int(m.group(1)) for m in (
-                re_runX.search(i.name) for i in paths
-            ) if m
+            int(m.group(1))
+            for m in (re_runX.search(i.name) for i in paths)
+            if m
         )
         last_run_num = max(run_numbers, default=0)
 
@@ -473,8 +485,8 @@ def parse_rm_dirs(rm_dirs: Iterable[str]) -> Set[str]:
             if os.path.isabs(part):
                 raise InputError("--rm option cannot take absolute paths")
             if (
-                part in {os.curdir, os.pardir} or
-                part.startswith(f"{os.pardir}{os.sep}")  # '../'
+                part in {os.curdir, os.pardir}
+                or part.startswith(f"{os.pardir}{os.sep}")  # '../'
             ):
                 raise InputError(
                     "--rm option cannot take paths that point to the "
@@ -504,8 +516,7 @@ def is_relative_to(path1: Union[Path, str], path2: Union[Path, str]) -> bool:
 
 
 def get_workflow_name_from_id(workflow_id: str) -> str:
-    """Workflow name is the ID shorn of the runN directory name.
-    """
+    """Workflow name is the ID shorn of the runN directory name."""
     cylc_run_dir = Path(get_cylc_run_dir())
     if Path(workflow_id).is_absolute():
         # this is a source directory, not an install dir: