X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=file_utils.py;h=7270e30b1fe1513746aaf986e87269968787e842;hb=c46018e2b1ddc78f9df557c3fb24d2c2c849f054;hp=464b0e76cfba0ef4e80ba5343c24bf433584b9b5;hpb=97fbe845e5dfdbda22521117c1783e1fd8515952;p=python_utils.git diff --git a/file_utils.py b/file_utils.py index 464b0e7..7270e30 100644 --- a/file_utils.py +++ b/file_utils.py @@ -7,11 +7,14 @@ import errno import hashlib import logging import os +import io import pathlib import time from typing import Optional import glob from os.path import isfile, join, exists +from uuid import uuid4 + logger = logging.getLogger(__name__) @@ -47,26 +50,48 @@ def create_path_if_not_exist(path, on_error=None): def does_file_exist(filename: str) -> bool: + """Returns True if a file exists and is a normal file. + + >>> does_file_exist(__file__) + True + """ return os.path.exists(filename) and os.path.isfile(filename) def does_directory_exist(dirname: str) -> bool: + """Returns True if a file exists and is a directory. + + >>> does_directory_exist('/tmp') + True + """ return os.path.exists(dirname) and os.path.isdir(dirname) def does_path_exist(pathname: str) -> bool: + """Just a more verbose wrapper around os.path.exists.""" return os.path.exists(pathname) def get_file_size(filename: str) -> int: + """Returns the size of a file in bytes.""" return os.path.getsize(filename) def is_normal_file(filename: str) -> bool: + """Returns True if filename is a normal file. + + >>> is_normal_file(__file__) + True + """ return os.path.isfile(filename) def is_directory(filename: str) -> bool: + """Returns True if filename is a directory. + + >>> is_directory('/tmp') + True + """ return os.path.isdir(filename) @@ -249,3 +274,30 @@ def get_files_recursive(directory: str): for subdir in get_directories(directory): for file_or_directory in get_files_recursive(subdir): yield file_or_directory + + +class FileWriter(object): + def __init__(self, filename: str) -> None: + self.filename = filename + uuid = uuid4() + self.tempfile = f'{filename}-{uuid}.tmp' + self.handle = None + + def __enter__(self) -> io.TextIOWrapper: + 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: + 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 + + +if __name__ == '__main__': + import doctest + doctest.testmod()