X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=src%2Fpyutils%2Ffiles%2Ffile_utils.py;h=027279fc03fc0bfdd1913b915a37ae93ecba480f;hb=HEAD;hp=a2f7baea2bd23ff43f583b8e7f63887794a6f1eb;hpb=77513ea630d72318684cf1d0a9198a22f4b547a7;p=pyutils.git diff --git a/src/pyutils/files/file_utils.py b/src/pyutils/files/file_utils.py index a2f7bae..027279f 100644 --- a/src/pyutils/files/file_utils.py +++ b/src/pyutils/files/file_utils.py @@ -21,7 +21,7 @@ import pathlib import re import time from os.path import exists, isfile, join -from typing import Callable, List, Literal, Optional, TextIO +from typing import IO, Any, Callable, List, Literal, Optional from uuid import uuid4 logger = logging.getLogger(__name__) @@ -49,7 +49,7 @@ def remove_hash_comments(x: str) -> str: def slurp_file( filename: str, *, - skip_blank_lines=False, + skip_blank_lines: bool = False, line_transformers: Optional[List[Callable[[str], str]]] = None, ): """Reads in a file's contents line-by-line to a memory buffer applying @@ -62,6 +62,9 @@ def slurp_file( Returns: A list of lines from the read and transformed file contents. + + Raises: + Exception: filename not found or can't be read. """ ret = [] @@ -88,6 +91,9 @@ def remove(path: str) -> None: Args: path: the path of the file to delete + Raises: + FileNotFoundError: the path to remove does not exist + >>> import os >>> filename = '/tmp/file_utils_test_file' >>> os.system(f'touch {filename}') @@ -97,6 +103,11 @@ def remove(path: str) -> None: >>> remove(filename) >>> does_file_exist(filename) False + + >>> remove("/tmp/23r23r23rwdfwfwefgdfgwerhwrgewrgergerg22r") + Traceback (most recent call last): + ... + FileNotFoundError: [Errno 2] No such file or directory: '/tmp/23r23r23rwdfwfwefgdfgwerhwrgewrgergerg22r' """ os.remove(path) @@ -292,26 +303,30 @@ def get_canonical_path(filespec: str) -> str: See also :meth:`get_path`, :meth:`without_path`. - >>> get_canonical_path('/home/scott/../../home/lynn/../scott/foo.txt') - '/usr/home/scott/foo.txt' + >>> get_canonical_path('./../../pyutils/files/file_utils.py') + '/usr/home/scott/lib/release/pyutils/files/file_utils.py' """ return os.path.realpath(filespec) -def create_path_if_not_exist(path, on_error=None) -> None: +def create_path_if_not_exist( + path: str, on_error: Callable[[str, OSError], None] = None +) -> None: """ Attempts to create path if it does not exist already. Args: path: the path to attempt to create - on_error: If True, it's invoked on error conditions. Otherwise - any exceptions are raised. + on_error: if provided, this is invoked on error conditions and + passed the path and OSError that it caused + + Raises: + OSError: an exception occurred and on_error was not set. See also :meth:`does_file_exist`. .. warning:: - Files are created with mode 0o0777 (i.e. world read/writeable). >>> import uuid @@ -347,6 +362,13 @@ def does_file_exist(filename: str) -> bool: Returns: True if filename exists and is a normal file. + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`create_path_if_not_exist`, :meth:`is_readable`. >>> does_file_exist(__file__) @@ -383,6 +405,13 @@ def is_writable(filename: str) -> bool: True if file exists, is a normal file and is writable by the current process. False otherwise. + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`is_readable`, :meth:`does_file_exist`. """ return os.access(filename, os.W_OK) @@ -398,6 +427,13 @@ def is_executable(filename: str) -> bool: True if file exists, is a normal file and is executable by the current process. False otherwise. + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`does_file_exist`, :meth:`is_readable`, :meth:`is_writable`. """ @@ -449,6 +485,13 @@ def is_normal_file(filename: str) -> bool: Returns: True if filename is a normal file. + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`is_directory`, :meth:`does_file_exist`, :meth:`is_symlink`. >>> is_normal_file(__file__) @@ -466,6 +509,13 @@ def is_directory(filename: str) -> bool: Returns: True if filename is a directory + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`does_directory_exist`, :meth:`is_normal_file`, :meth:`is_symlink`. @@ -484,6 +534,13 @@ def is_symlink(filename: str) -> bool: Returns: True if filename is a symlink, False otherwise. + .. note:: + A Python core philosophy is: it's easier to ask forgiveness + than permission (https://docs.python.org/3/glossary.html#term-EAFP). + That is, code that just tries an operation and handles the set of + Exceptions that may arise is the preferred style. That said, this + function can still be useful in some situations. + See also :meth:`is_directory`, :meth:`is_normal_file`. >>> is_symlink('/tmp') @@ -711,7 +768,8 @@ def set_file_raw_atime_and_mtime(filename: str, ts: float = None): def _convert_file_timestamp_to_datetime( - filename: str, producer + filename: str, + producer: Callable[[str], Optional[float]], ) -> Optional[datetime.datetime]: """ Converts a raw file timestamp into a Python datetime. @@ -719,7 +777,6 @@ def _convert_file_timestamp_to_datetime( Args: filename: file whose timestamps should be converted. producer: source of the timestamp. - Returns: The datetime. """ @@ -962,7 +1019,7 @@ def describe_file_timestamp(filename: str, extractor, *, brief=False) -> Optiona return describe_duration(age) -def describe_file_atime(filename: str, *, brief=False) -> Optional[str]: +def describe_file_atime(filename: str, *, brief: bool = False) -> Optional[str]: """ Describe how long ago a file was accessed. @@ -989,7 +1046,7 @@ def describe_file_atime(filename: str, *, brief=False) -> Optional[str]: return describe_file_timestamp(filename, lambda x: x.st_atime, brief=brief) -def describe_file_ctime(filename: str, *, brief=False) -> Optional[str]: +def describe_file_ctime(filename: str, *, brief: bool = False) -> Optional[str]: """Describes a file's creation time. Args: @@ -1013,7 +1070,7 @@ def describe_file_ctime(filename: str, *, brief=False) -> Optional[str]: return describe_file_timestamp(filename, lambda x: x.st_ctime, brief=brief) -def describe_file_mtime(filename: str, *, brief=False) -> Optional[str]: +def describe_file_mtime(filename: str, *, brief: bool = False) -> Optional[str]: """Describes how long ago a file was modified. Args: @@ -1095,13 +1152,13 @@ def get_files(directory: str): yield full_path -def get_matching_files(directory: str, glob: str): +def get_matching_files(directory: str, glob_string: str): """ Returns the subset of files whose name matches a glob. Args: directory: the directory to match files within. - glob: the globbing pattern (may include '*' and '?') to + glob_string: the globbing pattern (may include '*' and '?') to use when matching files. Returns: @@ -1111,7 +1168,7 @@ def get_matching_files(directory: str, glob: str): See also :meth:`get_files`, :meth:`expand_globs`. """ for filename in get_files(directory): - if fnmatch.fnmatch(filename, glob): + if fnmatch.fnmatch(filename, glob_string): yield filename @@ -1156,23 +1213,23 @@ def get_files_recursive(directory: str): yield file_or_directory -def get_matching_files_recursive(directory: str, glob: str): - """ - Returns the subset of files whose name matches a glob under a root recursively. +def get_matching_files_recursive(directory: str, glob_string: str): + """Returns the subset of files whose name matches a glob under a root recursively. Args: directory: the root under which to search - glob: a globbing pattern that describes the subset of files and directories - to return. May contain '?' and '*'. + glob_string: a globbing pattern that describes the subset of + files and directories to return. May contain '?' and '*'. Returns: A generator that yields all files and directories under the given root directory that match the given globbing pattern. See also :meth:`get_files_recursive`. + """ for filename in get_files_recursive(directory): - if fnmatch.fnmatch(filename, glob): + if fnmatch.fnmatch(filename, glob_string): yield filename @@ -1203,11 +1260,12 @@ class FileWriter(contextlib.AbstractContextManager): self.filename = filename uuid = uuid4() self.tempfile = f"{filename}-{uuid}.tmp" - self.handle: Optional[TextIO] = None + self.handle: Optional[IO[Any]] = None - def __enter__(self) -> TextIO: + def __enter__(self) -> IO[Any]: assert not does_path_exist(self.tempfile) self.handle = open(self.tempfile, mode="w") + assert self.handle return self.handle def __exit__(self, exc_type, exc_val, exc_tb) -> Literal[False]: @@ -1223,19 +1281,20 @@ class FileWriter(contextlib.AbstractContextManager): class CreateFileWithMode(contextlib.AbstractContextManager): """This helper context manager can be used instead of the typical pattern for creating a file if you want to ensure that the file - created is a particular permission mode upon creation. + created is a particular filesystem permission mode upon creation. Python's open doesn't support this; you need to set the os.umask and then create a descriptor to open via os.open, see below. >>> import os >>> filename = f'/tmp/CreateFileWithModeTest.{os.getpid()}' - >>> with CreateFileWithMode(filename, mode=0o600) as wf: + >>> with CreateFileWithMode(filename, filesystem_mode=0o600) as wf: ... print('This is a test', file=wf) >>> result = os.stat(filename) - Note: there is a high order bit set in this that is S_IFREG indicating - that the file is a "normal file". Clear it with the mask. + Note: there is a high order bit set in this that is S_IFREG + indicating that the file is a "normal file". Clear it with + the mask. >>> print(f'{result.st_mode & 0o7777:o}') 600 @@ -1244,14 +1303,22 @@ class CreateFileWithMode(contextlib.AbstractContextManager): >>> contents 'This is a test\\n' >>> remove(filename) + """ - def __init__(self, filename: str, mode: Optional[int] = 0o600) -> None: + def __init__( + self, + filename: str, + filesystem_mode: Optional[int] = 0o600, + open_mode: Optional[str] = "w", + ) -> None: """ Args: filename: path of the file to create. - mode: the UNIX-style octal mode with which to create the - filename. Defaults to 0o600. + filesystem_mode: the UNIX-style octal mode with which to create + the filename. Defaults to 0o600. + open_mode: the mode to use when opening the file (e.g. 'w', 'wb', + etc...) .. warning:: @@ -1259,20 +1326,25 @@ class CreateFileWithMode(contextlib.AbstractContextManager): """ self.filename = filename - if mode is not None: - self.mode = mode & 0o7777 + if filesystem_mode is not None: + self.filesystem_mode = filesystem_mode & 0o7777 + else: + self.filesystem_mode = 0o666 + if open_mode is not None: + self.open_mode = open_mode else: - self.mode = 0o666 - self.handle: Optional[TextIO] = None + self.open_mode = "w" + self.handle: Optional[IO[Any]] = None self.old_umask = os.umask(0) - def __enter__(self) -> TextIO: + def __enter__(self) -> IO[Any]: descriptor = os.open( path=self.filename, flags=(os.O_WRONLY | os.O_CREAT | os.O_TRUNC), - mode=self.mode, + mode=self.filesystem_mode, ) - self.handle = open(descriptor, "w") + self.handle = open(descriptor, self.open_mode) + assert self.handle return self.handle def __exit__(self, exc_type, exc_val, exc_tb) -> Literal[False]: