Merge pull request #314 from dclong/dev

Merge dev into main

Author: dclong

dsutil/__init__.py

@@ -3,4 +3,4 @@
 from . import git
 from . import poetry
 
-__version__ = "0.68.0"
+__version__ = "0.68.1"

dsutil/cv.py

@@ -42,7 +42,7 @@ def video_to_image(
 
 def resize_image(
     paths: Union[str, Path, Iterable[Path]], desdir: Union[str, Path, None],
-    size: tuple[int]
+    size: tuple[int, int]
 ) -> None:
     """Resize images to a given size.
 

dsutil/docker/builder.py

@@ -1,7 +1,7 @@
 """Docker related utils.
 """
 from __future__ import annotations
-from typing import Union
+from typing import Optional, Union
 from dataclasses import dataclass
 import tempfile
 from pathlib import Path
@@ -124,7 +124,7 @@ def __init__(
         git_url: str,
         branch: str = "dev",
         branch_fallback: str = "dev",
-        repo_path: dict[str, str] = None
+        repo_path: Optional[dict[str, str]] = None
     ):
         """Initialize a DockerImage object.
 
@@ -299,7 +299,7 @@ def build(
                 action="build",
                 seconds=(time.perf_counter_ns() - time_begin) / 1E9,
             )
-        except docker.errors.ImageNotFound as err:
+        except docker.errors.ImageNotFound:
             return DockerActionResult(
                 succeed=False,
                 err_msg="",
@@ -392,7 +392,7 @@ def __init__(
         builder: str = _get_docker_builder(),
     ):
         if isinstance(branch_urls, (str, Path)):
-            with open(branch_urls, "r") as fin:
+            with open(branch_urls, "r", encoding="utf-8") as fin:
                 branch_urls = yaml.load(fin, Loader=yaml.FullLoader)
         self._branch_urls = branch_urls
         self._branch_fallback = branch_fallback
@@ -514,7 +514,7 @@ def build_graph(self):
     def save_graph(self, output="graph.yaml") -> None:
         """Save the underlying graph structure to files.
         """
-        with open(output, "w") as fout:
+        with open(output, "w", encoding="utf-8") as fout:
             # nodes and attributes
             fout.write("nodes:\n")
             for node in self._graph.nodes:
@@ -536,7 +536,7 @@ def _login_servers(self) -> None:
 
     def build_images(
         self,
-        tag_build: str = None,
+        tag_build: Optional[str] = None,
         copy_ssh_to: str = "",
         push: bool = True,
         remove: bool = False,

dsutil/filesystem.py

@@ -2,7 +2,7 @@
 """Filesystem related util functions.
 """
 from __future__ import annotations
-from typing import Union, Iterable, Callable
+from typing import Optional, Union, Iterable, Callable
 import os
 import sys
 import re
@@ -20,7 +20,7 @@
 HOME = Path.home()
 
 
-def copy_if_exists(src: str, dst: str = HOME) -> bool:
+def copy_if_exists(src: str, dst: Union[str, Path] = HOME) -> bool:
     """Copy a file.
     No exception is thrown if the source file does not exist.
 
@@ -37,7 +37,9 @@ def copy_if_exists(src: str, dst: str = HOME) -> bool:
         return False
 
 
-def link_if_exists(src: str, dst: str = HOME, target_is_directory: bool = True) -> bool:
+def link_if_exists(
+    src: str, dst: Union[str, Path] = HOME, target_is_directory: bool = True
+) -> bool:
     """Make a symbolic link of a file.
     No exception is thrown if the source file does not exist.
 
@@ -217,7 +219,7 @@ def find_data_tables(
 def _find_data_tables_file(file, filter_, patterns) -> set[str]:
     if isinstance(file, str):
         file = Path(file)
-    text = file.read_text().lower()
+    text = file.read_text(encoding="utf-8").lower()
     patterns = {
         r"from\s+(\w+)\W*\s*",
         r"from\s+(\w+\.\w+)\W*\s*",
@@ -313,7 +315,7 @@ def find_ess_empty(path: Union[str, Path], ignore: Callable = _ignore) -> list[P
 
 
 def _find_ess_empty(
-    path: Path, ignore: Callable, ess_empty: dict[Path, bool], ess_empty_dir: list[str]
+    path: Path, ignore: Callable, ess_empty: dict[Path, bool], ess_empty_dir: list[Path]
 ):
     if is_ess_empty(path=path, ignore=ignore, ess_empty=ess_empty):
         ess_empty_dir.append(path)
@@ -326,7 +328,9 @@ def _find_ess_empty(
 
 
 def is_ess_empty(
-    path: Path, ignore: Callable = _ignore, ess_empty: dict[Path, bool] = None
+    path: Path,
+    ignore: Callable = _ignore,
+    ess_empty: Optional[dict[Path, bool]] = None
 ):
     """Check if a directory is essentially empty.
 
@@ -365,9 +369,9 @@ def is_ess_empty(
 
 def update_file(
     path: Path,
-    regex: list[tuple[str, str]] = None,
-    exact: list[tuple[str, str]] = None,
-    append: Union[str, Iterable[str]] = None,
+    regex: Optional[list[tuple[str, str]]] = None,
+    exact: Optional[list[tuple[str, str]]] = None,
+    append: Union[None, str, Iterable[str]] = None,
     exist_skip: bool = True,
 ) -> None:
     """Update a text file using regular expression substitution.
@@ -382,7 +386,7 @@ def update_file(
     """
     if isinstance(path, str):
         path = Path(path)
-    text = path.read_text()
+    text = path.read_text(encoding="utf-8")
     if regex:
         for pattern, replace in regex:
             text = re.sub(pattern, replace, text)
@@ -394,7 +398,7 @@ def update_file(
             append = "\n".join(append)
         if not exist_skip or append not in text:
             text += append
-    path.write_text(text)
+    path.write_text(text, encoding="utf-8")
 
 
 def get_files(dir_: Union[str, Path], exts: Union[str, list[str]]) -> Iterable[Path]:
@@ -581,7 +585,9 @@ def prune_json(input: Union[str, Path], output: Union[str, Path] = ""):
         else:
             output = input.with_name(input.stem + "_prune.json")
     skip = False
-    with input.open("r") as fin, output.open("w") as fout:
+    with input.open("r", encoding="utf-8") as fin, output.open(
+        "w", encoding="utf-8"
+    ) as fout:
         for line in fin:
             line = line.strip()
             if line == '"value_counts": {':
@@ -601,7 +607,7 @@ def _filter_num(path: Union[str, Path], pattern: str, num_lines: int):
     results = []
     res = []
     count = 0
-    for line in path.open():
+    for line in path.open(encoding="utf-8"):
         if count > 0:
             res.append(line)
             count -= 1
@@ -623,7 +629,7 @@ def _filter_sp(path: Union[str, Path], pattern: str, sub_pattern: str):
     results = []
     res = []
     sub = False
-    for line in path.open():
+    for line in path.open(encoding="utf-8"):
         if sub:
             if re.search(sub_pattern, line):
                 res.append(line)

dsutil/hadoop/kerberos.py

@@ -45,7 +45,7 @@ def save_passwd(passwd: str) -> None:
     """
     bytes_ = passwd.encode("ascii")
     encode = base64.b64encode(bytes_).decode()
-    with open(PROFILE, "w") as fout:
+    with open(PROFILE, "w", encoding="utf-8") as fout:
         fout.write(encode)
     os.chmod(PROFILE, 0o600)
 
@@ -58,7 +58,7 @@ def read_passwd() -> str:
     os.chmod(PROFILE, 0o600)
     if not os.path.isfile(PROFILE):
         return ""
-    with open(PROFILE, "r") as fin:
+    with open(PROFILE, "r", encoding="utf-8") as fin:
         return base64.b64decode(fin.read()).decode()
 
 
@@ -170,7 +170,7 @@ def _read_config(config: Union[Path, str]) -> Dict[str, Any]:
         config = Path(config)
     if not config.is_file():
         return {"email": {}}
-    with config.open("r") as fin:
+    with config.open("r", encoding="utf-8") as fin:
         return yaml.load(fin, Loader=yaml.FullLoader)
 
 

dsutil/hadoop/log.py

@@ -80,7 +80,7 @@ def __init__(
         log_file: Union[str, Path],
         context_size: int = 5,
         keywords: Sequence[str] = KEYWORDS,
-        patterns: Sequence[str] = PATTERNS,
+        patterns: Sequence[tuple[str, str]] = PATTERNS,
         output: Union[str, Path] = "",
         threshold: float = 0.7,
         dump_by_keyword: bool = False,
@@ -89,11 +89,11 @@ def __init__(
                           if isinstance(log_file, Path) else Path(log_file)).resolve()
         self._context_size: int = context_size
         self._keywords: Sequence[str] = keywords
-        self._patterns: Sequence[str] = patterns
+        self._patterns: Sequence[tuple[str, str]] = patterns
         self._num_rows: int = 0
         self._lookup: dict[str, dict[str, int]] = {kwd: {} for kwd in self._keywords}
         self._queue: deque = deque()
-        self._output: str = self._get_output(output)
+        self._output: Path = self._get_output(output)
         self._threshold: float = threshold
         self._dump_by_keyword: bool = dump_by_keyword
 
@@ -153,15 +153,15 @@ def _count_rows(self):
         if self._num_rows:
             return
         logger.info("Counting total number of rows ...")
-        with open(self._log_file, "r") as fin:
+        with open(self._log_file, "r", encoding="utf-8") as fin:
             self._num_rows = sum(1 for line in fin)
         logger.info("Total number of rows: {:,}", self._num_rows)
 
     def _scan_error_lines(self) -> None:
         print()
         logger.info("Scanning for error lines in the log ...")
         lines = [DASH_50 + " Possible Error Lines " + DASH_50 + "\n"]
-        with open(self._log_file, "r") as fin:
+        with open(self._log_file, "r", encoding="utf-8") as fin:
             dump_flag = -1
             for idx, line in tqdm(enumerate(fin), total=self._num_rows):
                 self._queue.append(f"L{idx}: {line}")
@@ -179,7 +179,7 @@ def _scan_error_lines(self) -> None:
                     dump_flag = -1
             if dump_flag >= 0:
                 self._dump_queue(lines)
-        with open(self._output, "w") as fout:
+        with open(self._output, "w", encoding="utf-8") as fout:
             fout.writelines(lines)
         logger.info("Possible Error Lines have been dumped into {}", self._output)
 
@@ -210,7 +210,7 @@ def _dedup_log(self):
             lines_unique.extend(self._dedup_log_1(kwd, lines, dir_))
         lines_unique = [self._error_priority(line) for line in lines_unique]
         lines_unique.sort()
-        with self._output.open("a") as fout:
+        with self._output.open("a", encoding="utf-8") as fout:
             self._write_lines_unique(lines_unique, fout)
         self._write_lines_unique(lines_unique, sys.stdout)
 

dsutil/hadoop/pyspark_submit.py

@@ -135,23 +135,23 @@ def submit(self, cmd: str, attachments: Union[None, list[str]] = None) -> bool:
         logger.info("Submitting Spark job ...\n{}", cmd)
         stdout = []
         self._spark_submit_log.clear()
-        process = sp.Popen(cmd, shell=True, stderr=sp.PIPE)
-        while True:
-            if process.poll() is None:
-                line = process.stderr.readline().decode().rstrip()  # pytype: disable=attribute-error
-                line = self._filter(line, time_begin, self._spark_log_filter)
-                if line:
-                    print(line)
-                    stdout.append(line)
-            else:
-                for line in process.stderr.readlines():  # pytype: disable=attribute-error
-                    line = self._filter(
-                        line.decode().rstrip(), time_begin, self._spark_log_filter
-                    )
+        with sp.Popen(cmd, shell=True, stderr=sp.PIPE) as process:
+            while True:
+                if process.poll() is None:
+                    line = process.stderr.readline().decode().rstrip()  # pytype: disable=attribute-error
+                    line = self._filter(line, time_begin, self._spark_log_filter)
                     if line:
                         print(line)
                         stdout.append(line)
-                break
+                else:
+                    for line in process.stderr.readlines():  # pytype: disable=attribute-error
+                        line = self._filter(
+                            line.decode().rstrip(), time_begin, self._spark_log_filter
+                        )
+                        if line:
+                            print(line)
+                            stdout.append(line)
+                    break
         # status
         status = self._final_status(stdout)
         app_id = self._app_id(stdout)
@@ -376,7 +376,7 @@ def submit(args: Namespace) -> None:
     if not args.config:
         config = {}
     else:
-        with open(args.config, "r") as fin:
+        with open(args.config, "r", encoding="utf-8") as fin:
             config = yaml.load(fin, Loader=yaml.FullLoader)
     # handle various options
     if args.spark_submit_local:

dsutil/hash.py

@@ -23,12 +23,12 @@ def rmd5(path: Union[str, Path], output: Union[str, Path] = "") -> str:
     if output:
         if isinstance(output, str):
             output = Path(output)
-        with output.open("w") as fout:
+        with output.open("w", encoding="utf-8") as fout:
             fout.write(text)
     return hashlib.md5(text.encode()).hexdigest()
 
 
-def _rmd5(path: Path, res: list[tuple[str]]) -> None:
+def _rmd5(path: Path, res: list[str]) -> None:
     """Helper function of rmd5.
 
     :param path: The Path object of a file or directory.

dsutil/poetry.py

@@ -1,7 +1,7 @@
 """This module makes it easy to work with poetry to managing your Python project.
 """
 from __future__ import annotations
-from typing import Union, Iterable
+from typing import Optional, Union, Iterable
 from pathlib import Path
 import sys
 import os
@@ -101,7 +101,7 @@ def _update_version(ver: str, proj_dir: Path) -> None:
 def version(
     ver: str = "",
     commit: bool = False,
-    proj_dir: Path = None,
+    proj_dir: Optional[Path] = None,
 ) -> None:
     """List or update the version of the package.
 
@@ -166,7 +166,7 @@ def add_tag_release(
 def format_code(
     inplace: bool = False,
     commit: bool = False,
-    proj_dir: Path = None,
+    proj_dir: Optional[Path] = None,
     files: Iterable[Union[Path, str]] = ()
 ) -> None:
     """Format code.
@@ -304,7 +304,9 @@ def build_package(
     sp.run(f"cd '{proj_dir}' && poetry build", shell=True, check=True)
 
 
-def clean(proj_dir: Path = None, ignore: Union[str, Path, None] = None) -> None:
+def clean(
+    proj_dir: Optional[Path] = None, ignore: Union[str, Path, None] = None
+) -> None:
     """Remove non-essential files from the current project.
 
     :param proj_dir: The root directory of the Poetry project.
@@ -319,7 +321,7 @@ def clean(proj_dir: Path = None, ignore: Union[str, Path, None] = None) -> None:
     if not ignore.is_file():
         return
     logger.info("Use the GitIgnore file: {}", ignore)
-    with ignore.open("r") as fin:
+    with ignore.open("r", encoding="utf-8") as fin:
         patterns = [line.strip() for line in fin]
     spec = pathspec.PathSpec.from_lines(pathspec.patterns.GitWildMatchPattern, patterns)
     _clean(proj_dir, spec)