X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=file_utils.py;h=f273ea4f9a5fe08da8c1f15e3108c65b7c33d0ea;hb=a4bf4d05230474ad14243d67ac7f8c938f670e58;hp=9fa8fd209264ec44378adff464904c8bc900b10d;hpb=eedcbd4f64af13ec2098508c3d839a60f7e9ffce;p=python_utils.git diff --git a/file_utils.py b/file_utils.py index 9fa8fd2..f273ea4 100644 --- a/file_utils.py +++ b/file_utils.py @@ -2,7 +2,6 @@ """Utilities for working with files.""" -from dataclasses import dataclass import datetime import errno import hashlib @@ -10,17 +9,49 @@ import logging import os import io import pathlib +import re import time from typing import Optional import glob from os.path import isfile, join, exists -from typing import List +from typing import List, TextIO from uuid import uuid4 logger = logging.getLogger(__name__) +def remove_newlines(x): + return x.replace('\n', '') + + +def strip_whitespace(x): + return x.strip() + + +def remove_hash_comments(x): + return re.sub(r'#.*$', '', x) + + +def slurp_file( + filename: str, + *, + skip_blank_lines=False, + line_transformers=[], +): + 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 skip_blank_lines and line == '': + continue + ret.append(line) + return ret + + def remove(path: str) -> None: """Deletes a file. Raises if path refers to a directory or a file that doesn't exist. @@ -185,6 +216,18 @@ def does_file_exist(filename: str) -> bool: return os.path.exists(filename) and os.path.isfile(filename) +def file_is_readable(filename: str) -> bool: + return does_file_exist(filename) and os.access(filename, os.R_OK) + + +def file_is_writable(filename: str) -> bool: + return does_file_exist(filename) and os.access(filename, os.W_OK) + + +def file_is_executable(filename: str) -> bool: + return does_file_exist(filename) and os.access(filename, os.X_OK) + + def does_directory_exist(dirname: str) -> bool: """Returns True if a file exists and is a directory. @@ -289,11 +332,13 @@ def get_file_md5(filename: str) -> str: def set_file_raw_atime(filename: str, atime: float): mtime = get_file_raw_mtime(filename) + assert mtime os.utime(filename, (atime, mtime)) def set_file_raw_mtime(filename: str, mtime: float): atime = get_file_raw_atime(filename) + assert atime os.utime(filename, (atime, mtime)) @@ -391,8 +436,8 @@ def describe_file_mtime(filename: str, *, brief=False) -> Optional[str]: return describe_file_timestamp(filename, lambda x: x.st_mtime, brief=brief) -def touch_file(filename: str, *, mode: Optional[int] = 0o666) -> bool: - return pathlib.Path(filename, mode=mode).touch() +def touch_file(filename: str, *, mode: Optional[int] = 0o666): + pathlib.Path(filename, mode=mode).touch() def expand_globs(in_filename: str): @@ -427,14 +472,14 @@ class FileWriter(object): self.filename = filename uuid = uuid4() self.tempfile = f'{filename}-{uuid}.tmp' - self.handle = None + self.handle: Optional[TextIO] = None - def __enter__(self) -> io.TextIOWrapper: + def __enter__(self) -> TextIO: assert not does_path_exist(self.tempfile) self.handle = open(self.tempfile, mode="w") return self.handle - def __exit__(self, exc_type, exc_val, exc_tb) -> bool: + def __exit__(self, exc_type, exc_val, exc_tb) -> Optional[bool]: if self.handle is not None: self.handle.close() cmd = f'/bin/mv -f {self.tempfile} {self.filename}'