X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=file_utils.py;h=91aeea072b03d94d670a13ca1a348b407d5734b8;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=d1e2eff1fac729b6251179ad7dfc584a0674215f;hpb=0d63d44ac89aab38fe95f36497adaf95110ab949;p=python_utils.git diff --git a/file_utils.py b/file_utils.py index d1e2eff..91aeea0 100644 --- a/file_utils.py +++ b/file_utils.py @@ -1,7 +1,10 @@ #!/usr/bin/env python3 +# © Copyright 2021-2022, Scott Gasch + """Utilities for working with files.""" +import contextlib import datetime import errno import glob @@ -12,21 +15,21 @@ import pathlib import re import time from os.path import exists, isfile, join -from typing import List, Optional, TextIO +from typing import Callable, List, Literal, Optional, TextIO from uuid import uuid4 logger = logging.getLogger(__name__) -def remove_newlines(x): +def remove_newlines(x: str) -> str: return x.replace('\n', '') -def strip_whitespace(x): +def strip_whitespace(x: str) -> str: return x.strip() -def remove_hash_comments(x): +def remove_hash_comments(x: str) -> str: return re.sub(r'#.*$', '', x) @@ -34,15 +37,16 @@ def slurp_file( filename: str, *, skip_blank_lines=False, - line_transformers=[], + line_transformers: Optional[List[Callable[[str], str]]] = None, ): ret = [] if not file_is_readable(filename): raise Exception(f'{filename} can\'t be read.') with open(filename) as rf: for line in rf: - for transformation in line_transformers: - line = transformation(line) + if line_transformers is not None: + for transformation in line_transformers: + line = transformation(line) if skip_blank_lines and line == '': continue ret.append(line) @@ -460,7 +464,7 @@ def get_files_recursive(directory: str): yield file_or_directory -class FileWriter(object): +class FileWriter(contextlib.AbstractContextManager): """A helper that writes a file to a temporary location and then moves it atomically to its ultimate destination on close. @@ -477,14 +481,14 @@ class FileWriter(object): self.handle = open(self.tempfile, mode="w") return self.handle - def __exit__(self, exc_type, exc_val, exc_tb) -> Optional[bool]: + def __exit__(self, exc_type, exc_val, exc_tb) -> Literal[False]: if self.handle is not None: self.handle.close() cmd = f'/bin/mv -f {self.tempfile} {self.filename}' ret = os.system(cmd) if (ret >> 8) != 0: - raise Exception(f'{cmd} failed, exit value {ret>>8}') - return None + raise Exception(f'{cmd} failed, exit value {ret>>8}!') + return False if __name__ == '__main__':