Make unscrambler use a cache per instance instead of module and

[python_utils.git] / file_utils.py

diff --git a/file_utils.py b/file_utils.py
index 464b0e76cfba0ef4e80ba5343c24bf433584b9b5..9fa8fd209264ec44378adff464904c8bc900b10d 100644 (file)
--- a/file_utils.py
+++ b/file_utils.py
@@ -2,20 +2,150 @@
 
 """Utilities for working with files."""
 
 
 """Utilities for working with files."""
 

+from dataclasses import dataclass

 import datetime
 import errno
 import hashlib
 import logging
 import os
 import datetime
 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
 import pathlib
 import time
 from typing import Optional
 import glob
 from os.path import isfile, join, exists

+from typing import List
+from uuid import uuid4
+

 
 logger = logging.getLogger(__name__)
 
 
 
 logger = logging.getLogger(__name__)
 
 

+def remove(path: str) -> None:
+    """Deletes a file.  Raises if path refers to a directory or a file
+    that doesn't exist.
+
+    >>> import os
+    >>> filename = '/tmp/file_utils_test_file'
+    >>> os.system(f'touch {filename}')
+    0
+    >>> does_file_exist(filename)
+    True
+    >>> remove(filename)
+    >>> does_file_exist(filename)
+    False
+
+    """
+    os.remove(path)
+
+
+def delete(path: str) -> None:
+    os.remove(path)
+
+
+def without_extension(path: str) -> str:
+    """Remove one extension from a file or path.
+
+    >>> without_extension('foobar.txt')
+    'foobar'
+
+    >>> without_extension('/home/scott/frapp.py')
+    '/home/scott/frapp'
+
+    >>> without_extension('a.b.c.tar.gz')
+    'a.b.c.tar'
+
+    >>> without_extension('foobar')
+    'foobar'
+
+    """
+    return os.path.splitext(path)[0]
+
+
+def without_all_extensions(path: str) -> str:
+    """Removes all extensions from a path; handles multiple extensions
+    like foobar.tar.gz -> foobar.
+
+    >>> without_all_extensions('/home/scott/foobar.1.tar.gz')
+    '/home/scott/foobar'
+
+    """
+    while '.' in path:
+        path = without_extension(path)
+    return path
+
+
+def get_extension(path: str) -> str:
+    """Extract and return one extension from a file or path.
+
+    >>> get_extension('this_is_a_test.txt')
+    '.txt'
+
+    >>> get_extension('/home/scott/test.py')
+    '.py'
+
+    >>> get_extension('foobar')
+    ''
+
+    """
+    return os.path.splitext(path)[1]
+
+
+def get_all_extensions(path: str) -> List[str]:
+    """Return the extensions of a file or path in order.
+
+    >>> get_all_extensions('/home/scott/foo.tar.gz.1')
+    ['.tar', '.gz', '.1']
+
+    """
+    ret = []
+    while True:
+        ext = get_extension(path)
+        path = without_extension(path)
+        if ext:
+            ret.append(ext)
+        else:
+            ret.reverse()
+            return ret
+
+
+def without_path(filespec: str) -> str:
+    """Returns the base filename without any leading path.
+
+    >>> without_path('/home/scott/foo.py')
+    'foo.py'
+
+    >>> without_path('foo.py')
+    'foo.py'
+
+    """
+    return os.path.split(filespec)[1]
+
+
+def get_path(filespec: str) -> str:
+    """Returns just the path of the filespec by removing the filename and
+    extension.
+
+    >>> get_path('/home/scott/foobar.py')
+    '/home/scott'
+
+    >>> get_path('~scott/frapp.txt')
+    '~scott'
+
+    """
+    return os.path.split(filespec)[0]
+
+
+def get_canonical_path(filespec: str) -> str:
+    """Returns a canonicalized absolute path.
+
+    >>> get_canonical_path('/home/scott/../../home/lynn/../scott/foo.txt')
+    '/usr/home/scott/foo.txt'
+
+    """
+    return os.path.realpath(filespec)
+
+

 def create_path_if_not_exist(path, on_error=None):
     """
     Attempts to create path if it does not exist. If on_error is
 def create_path_if_not_exist(path, on_error=None):
     """
     Attempts to create path if it does not exist. If on_error is


@@ -47,38 +177,79 @@ def create_path_if_not_exist(path, on_error=None):
 
 
 def does_file_exist(filename: str) -> bool:
 
 
 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:
     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:
     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:
     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:
     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:
     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)
 
 
 def is_symlink(filename: str) -> bool:
     return os.path.isdir(filename)
 
 
 def is_symlink(filename: str) -> bool:

+    """True if filename is a symlink, False otherwise.
+
+    >>> is_symlink('/tmp')
+    False
+
+    >>> is_symlink('/home')
+    True
+
+    """

     return os.path.islink(filename)
 
 
 def is_same_file(file1: str, file2: str) -> bool:
     return os.path.islink(filename)
 
 
 def is_same_file(file1: str, file2: str) -> bool:

+    """Returns True if the two files are the same inode.
+
+    >>> is_same_file('/tmp', '/tmp/../tmp')
+    True
+
+    >>> is_same_file('/tmp', '/home')
+    False
+
+    """

     return os.path.samefile(file1, file2)
 
 
 def get_file_raw_timestamps(filename: str) -> Optional[os.stat_result]:
     return os.path.samefile(file1, file2)
 
 
 def get_file_raw_timestamps(filename: str) -> Optional[os.stat_result]:

+    """Stats the file and returns an os.stat_result or None on error."""

     try:
         return os.stat(filename)
     except Exception as e:
     try:
         return os.stat(filename)
     except Exception as e:


@@ -106,6 +277,7 @@ def get_file_raw_ctime(filename: str) -> Optional[float]:
 
 
 def get_file_md5(filename: str) -> str:
 
 
 def get_file_md5(filename: str) -> str:

+    """Hashes filename's contents and returns an MD5."""

     file_hash = hashlib.md5()
     with open(filename, "rb") as f:
         chunk = f.read(8192)
     file_hash = hashlib.md5()
     with open(filename, "rb") as f:
         chunk = f.read(8192)


@@ -195,10 +367,9 @@ def get_file_mtime_timedelta(filename: str) -> Optional[datetime.timedelta]:
     return get_file_timestamp_timedelta(filename, lambda x: x.st_mtime)
 
 
     return get_file_timestamp_timedelta(filename, lambda x: x.st_mtime)
 
 

-def describe_file_timestamp(
-    filename: str, extractor, *, brief=False
-) -> Optional[str]:
+def describe_file_timestamp(filename: str, extractor, *, brief=False) -> Optional[str]:

     from datetime_utils import describe_duration, describe_duration_briefly
     from datetime_utils import describe_duration, describe_duration_briefly

+

     age = get_file_timestamp_age_seconds(filename, extractor)
     if age is None:
         return None
     age = get_file_timestamp_age_seconds(filename, extractor)
     if age is None:
         return None


@@ -220,8 +391,8 @@ def describe_file_mtime(filename: str, *, brief=False) -> Optional[str]:
     return describe_file_timestamp(filename, lambda x: x.st_mtime, brief=brief)
 
 
     return describe_file_timestamp(filename, lambda x: x.st_mtime, brief=brief)
 
 

-def touch_file(filename: str) -> bool:
-    return pathlib.Path(filename).touch()
+def touch_file(filename: str, *, mode: Optional[int] = 0o666) -> bool:
+    return pathlib.Path(filename, mode=mode).touch()

 
 
 def expand_globs(in_filename: str):
 
 
 def expand_globs(in_filename: str):


@@ -249,3 +420,31 @@ 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
     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()