Commit 493bcedb by Marten Ringwelski

Use pathlib

As pathlib.Path.__init__ accepts both a string and Path-like objects we can
stay backward compatible and still accept strings.

Also Two functions are deprecated since they don't provide much readability
over the pathlib implementation.
parent 86686b18
from .fail_safe_file_operations import (
get_binary_from_file, get_string_list_from_file, write_binary_to_file, get_safe_name, delete_file, get_files_in_dir,
get_dirs_in_dir, create_symlink, get_dir_of_file, safe_rglob
)
from .file_functions import read_in_chunks, get_directory_for_filename, create_dir_for_file, human_readable_file_size
from .fail_safe_file_operations import (create_symlink, delete_file,
get_binary_from_file, get_dir_of_file,
get_dirs_in_dir, get_files_in_dir,
get_safe_name,
get_string_list_from_file, safe_rglob,
write_binary_to_file)
from .file_functions import (create_dir_for_file, get_directory_for_filename,
human_readable_file_size, read_in_chunks)
from .git_functions import get_version_string_from_git
__all__ = [
......
......@@ -3,88 +3,73 @@ import os
import re
import sys
from pathlib import Path
from typing import Iterable
from typing import Iterable, List, Union
from .file_functions import create_dir_for_file
def get_binary_from_file(file_path):
def get_binary_from_file(file_path: Union[str, Path]) -> Union[str, bytes]:
'''
Fail-safe file read operation. Symbolic links are converted to text files including the link.
Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:return: file's binary as bytes; returns empty byte string on error
'''
try:
if os.path.islink(file_path):
binary = 'symbolic link -> {}'.format(os.readlink(file_path))
p = Path(file_path)
if p.is_symlink():
# We need to wait for python 3.9 for Path.readlink
binary = f'symbolic link -> {os.readlink(p)}'
else:
with open(file_path, 'rb') as f:
binary = f.read()
binary = p.read_bytes()
except Exception as e:
logging.error('Could not read file: {} {}'.format(sys.exc_info()[0].__name__, e))
logging.error(f'Could not read file: {sys.exc_info()[0].__name__} {e}')
binary = b''
return binary
def get_string_list_from_file(file_path):
def get_string_list_from_file(file_path: Union[str, Path]) -> List[str]:
'''
Fail-safe file read operation returning a list of text strings.
Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:return: file's content as text string list; returns empty list on error
'''
try:
raw = get_binary_from_file(file_path)
raw_string = raw.decode(encoding='utf-8', errors='replace')
cleaned_string = _rm_cr(raw_string)
return cleaned_string.split('\n')
except Exception as e:
logging.error('Could not read file: {} {}'.format(sys.exc_info()[0].__name__, e))
return []
raw = get_binary_from_file(file_path)
raw_string = raw.decode(encoding='utf-8', errors='replace')
cleaned_string = _rm_cr(raw_string)
return cleaned_string.split('\n')
def _rm_cr(input_string):
return input_string.replace('\r', '')
def write_binary_to_file(file_binary, file_path, overwrite=False, file_copy=False):
def write_binary_to_file(file_binary: Union[str, bytes], file_path: Union[str, Path], overwrite: bool = False, file_copy: bool = False) -> None:
'''
Fail-safe file write operation. Creates directories if needed.
Errors are logged. No exception raised.
:param file_binary: binary to write into the file
:type file_binary: bytes or str
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:param file_path_str: Path of the file. Can be absolute or relative to the current directory.
:param overwrite: overwrite file if it exists
:type overwrite: bool
:default overwrite: False
:param file_copy: If overwrite is false and file already exists, write into new file and add a counter to the file name.
:type file_copy: bool
:default file_copy: False
:return: None
'''
try:
file_path = Path(file_path)
create_dir_for_file(file_path)
if not os.path.exists(file_path) or overwrite:
_write_file(file_path, file_binary)
elif file_copy and not overwrite:
new_path = _get_counted_file_path(file_path)
_write_file(new_path, file_binary)
if file_path.exists() and (not overwrite or file_copy):
file_path = Path(_get_counted_file_path(str(file_path)))
file_path.write_bytes(file_binary)
except Exception as e:
logging.error('Could not write file: {} {}'.format(sys.exc_info()[0].__name__, e))
def _write_file(file_path, binary):
with open(file_path, 'wb') as f:
f.write(binary)
def _get_counted_file_path(original_path):
tmp = re.search(r'-([0-9]+)\Z', original_path)
if tmp is not None:
......@@ -95,55 +80,46 @@ def _get_counted_file_path(original_path):
return new_file_path
def delete_file(file_path):
def delete_file(file_path: Union[str, Path]) -> None:
'''
Fail-safe delete file operation. Deletes a file if it exists.
Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:return: None
'''
try:
os.unlink(file_path)
Path(file_path).unlink()
except Exception as e:
logging.error('Could not delete file: {} {}'.format(sys.exc_info()[0].__name__, e))
def create_symlink(src_path, dst_path):
def create_symlink(src_path: Union[str, Path], dst_path: Union[str, Path]) -> None:
'''
Fail-safe symlink operation. Symlinks a file if dest does not exist.
Errors are logged. No exception raised.
:param src_path: src file
:type src_path: str
:param dst_path: link location
:type dst_path: str
:return: None
'''
try:
create_dir_for_file(dst_path)
os.symlink(src_path, dst_path)
Path(dst_path).symlink_to(src_path)
except FileExistsError as e:
logging.debug('Could not create Link: File exists: {}'.format(e))
except Exception as e:
logging.error('Could not create link: {} {}'.format(sys.exc_info()[0].__name__, e))
def get_safe_name(file_name, max_size=200, valid_characters='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_+. '):
def get_safe_name(file_name: str, max_size: int = 200, valid_characters: str = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_+. ') -> str:
'''
removes all problematic characters from a file name
cuts file names if they are too long
:param file_name: Original file name
:type file_name: str
:param max_size: maximum allowed file name length
:type max_size: int
:default max_size: 200
:param valid_characters: characters that shall be allowed in a file name
:type valid_characters: str
:default valid_characters: 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_+. '
:return: str
'''
allowed_charachters = set(valid_characters)
safe_name = filter(lambda x: x in allowed_charachters, file_name)
......@@ -154,54 +130,49 @@ def get_safe_name(file_name, max_size=200, valid_characters='abcdefghijklmnopqrs
return safe_name
def get_files_in_dir(directory_path):
def get_files_in_dir(directory_path: Union[str, Path]) -> List[str]:
'''
Returns a list with the absolute paths of all files in the directory directory_path
:param directory_path: directory including files
:type directory_path: str
:return: list
'''
result = []
try:
for file_path, _, files in os.walk(directory_path):
for file_ in files:
result.append(os.path.abspath(os.path.join(file_path, file_)))
result.append(str(Path(file_path, file_).absolute()))
except Exception as e:
logging.error('Could not get files: {} {}'.format(sys.exc_info()[0].__name__, e))
return result
def get_dirs_in_dir(directory_path):
def get_dirs_in_dir(directory_path: Union[str, Path]) -> List[str]:
'''
Returns a list with the absolute paths of all 1st level sub-directories in the directory directory_path.
:param directory_path: directory including sub-directories
:type directory_path: str
:return: list
'''
result = []
try:
dir_content = os.listdir(directory_path)
for item in dir_content:
dir_path = os.path.join(directory_path, item)
if os.path.isdir(dir_path):
result.append(dir_path)
p = Path(directory_path)
for item in p.iterdir():
if not Path(item).is_dir():
continue
result.append(str(item.resolve()))
except Exception as e:
logging.error('Could not get directories: {} {}'.format(sys.exc_info()[0].__name__, e))
return result
def get_dir_of_file(file_path):
def get_dir_of_file(file_path: Union[str, Path]) -> str:
'''
Returns absolute path of the directory including file
:param file_path: Path of the file
:type: path-like object
:return: string
'''
try:
return os.path.dirname(os.path.abspath(file_path))
return str(Path(file_path).resolve().parent)
except Exception as e:
logging.error('Could not get directory path: {} {}'.format(sys.exc_info()[0].__name__, e))
return '/'
......
import os
import io
from pathlib import Path
from typing import Type, Union
import bitmath
def read_in_chunks(file_object, chunk_size=1024):
def read_in_chunks(file_object: Type[io.BufferedReader], chunk_size=1024) -> bytes:
'''
Helper function to read large file objects iteratively in smaller chunks. Can be used like this::
......@@ -12,7 +15,6 @@ def read_in_chunks(file_object, chunk_size=1024):
:param file_object: The file object from which the chunk data is read. Must be a subclass of ``io.BufferedReader``.
:param chunk_size: Number of bytes to read per chunk.
:type chunk_size: int
:return: Returns a generator to iterate over all chunks, see above for usage.
'''
while True:
......@@ -22,35 +24,36 @@ def read_in_chunks(file_object, chunk_size=1024):
yield data
def get_directory_for_filename(filename):
def get_directory_for_filename(filename: Union[str, Path]) -> str:
'''
Convenience function which returns the absolute path to the directory that contains the given file name.
:param filename: Path of the file. Can be absolute or relative to the current directory.
:type filename: str
:return: Absolute path of the directory
.. deprecated::
You should use pathlib instead of this function.
'''
return os.path.dirname(os.path.abspath(filename))
return str(Path(filename).resolve().parent)
def create_dir_for_file(file_path):
def create_dir_for_file(file_path: Union[str, Path]) -> None:
'''
Creates all directories of file path. File path may include the file as well.
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:return: None
.. deprecated::
You should use pathlib instead of this function.
'''
directory = os.path.dirname(os.path.abspath(file_path))
os.makedirs(directory, exist_ok=True)
Path(file_path).resolve().parent.mkdir(parents=True, exist_ok=True)
def human_readable_file_size(size_in_bytes):
def human_readable_file_size(size_in_bytes: int) -> None:
'''
Returns a nicly human readable file size
:param size_in_bytes: Size in Bytes
:type size_in_bytes: int
:return: str
'''
return bitmath.Byte(bytes=size_in_bytes).best_prefix().format('{value:.2f} {unit}')
import subprocess
def get_version_string_from_git(directory_name):
def get_version_string_from_git(directory_name: str) -> str:
return subprocess.check_output(['git', 'describe', '--always'], cwd=directory_name).strip().decode('utf-8')
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment