Unverified Commit 22f1745e by Jörg Stucke Committed by GitHub

Merge pull request #6 from maringuu/pathlib

Use pathlib
parents 86686b18 3909ba50
from .fail_safe_file_operations import ( from .fail_safe_file_operations import (create_symlink, delete_file,
get_binary_from_file, get_string_list_from_file, write_binary_to_file, get_safe_name, delete_file, get_files_in_dir, get_binary_from_file, get_dir_of_file,
get_dirs_in_dir, create_symlink, get_dir_of_file, safe_rglob get_dirs_in_dir, get_files_in_dir,
) get_safe_name,
from .file_functions import read_in_chunks, get_directory_for_filename, create_dir_for_file, human_readable_file_size 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 from .git_functions import get_version_string_from_git
__all__ = [ __all__ = [
......
...@@ -3,86 +3,67 @@ import os ...@@ -3,86 +3,67 @@ import os
import re import re
import sys import sys
from pathlib import Path from pathlib import Path
from typing import Iterable from typing import Iterable, List, Union
from .file_functions import create_dir_for_file 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. Fail-safe file read operation. Symbolic links are converted to text files including the link.
Errors are logged. No exception raised. Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory. :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 :return: file's binary as bytes; returns empty byte string on error
''' '''
try: try:
if os.path.islink(file_path): path = Path(file_path)
binary = 'symbolic link -> {}'.format(os.readlink(file_path)) if path.is_symlink():
# We need to wait for python 3.9 for Path.readlink
binary = f'symbolic link -> {os.readlink(path)}'
else: else:
with open(file_path, 'rb') as f: binary = path.read_bytes()
binary = f.read()
except Exception as e: except Exception as e:
logging.error('Could not read file: {} {}'.format(sys.exc_info()[0].__name__, e)) logging.error(f'Could not read file: {e}', exc_info=True)
binary = b'' binary = b''
return binary 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. Fail-safe file read operation returning a list of text strings.
Errors are logged. No exception raised. Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory. :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 :return: file's content as text string list; returns empty list on error
''' '''
try: raw = get_binary_from_file(file_path)
raw = get_binary_from_file(file_path) raw_string = raw.decode(encoding='utf-8', errors='replace')
raw_string = raw.decode(encoding='utf-8', errors='replace') cleaned_string = raw_string.replace('\r', '')
cleaned_string = _rm_cr(raw_string) return cleaned_string.split('\n')
return cleaned_string.split('\n')
except Exception as e:
logging.error('Could not read file: {} {}'.format(sys.exc_info()[0].__name__, e))
return []
def _rm_cr(input_string):
return input_string.replace('\r', '')
def write_binary_to_file(file_binary: Union[str, bytes], file_path: Union[str, Path], overwrite: bool = False, file_copy: bool = False) -> None:
def write_binary_to_file(file_binary, file_path, overwrite=False, file_copy=False):
''' '''
Fail-safe file write operation. Creates directories if needed. Fail-safe file write operation. Creates directories if needed.
Errors are logged. No exception raised. Errors are logged. No exception raised.
:param file_binary: binary to write into the file :param file_binary: binary to write into the file
:type file_binary: bytes or str :param file_path_str: Path of the file. Can be absolute or relative to the current directory.
:param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:param overwrite: overwrite file if it exists :param overwrite: overwrite file if it exists
:type overwrite: bool
:default overwrite: False :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. :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 :default file_copy: False
:return: None
''' '''
try: try:
file_path = Path(file_path)
create_dir_for_file(file_path) create_dir_for_file(file_path)
if not os.path.exists(file_path) or overwrite: if file_path.exists() and (not overwrite or file_copy):
_write_file(file_path, file_binary) file_path = Path(_get_counted_file_path(str(file_path)))
elif file_copy and not overwrite: file_path.write_bytes(file_binary)
new_path = _get_counted_file_path(file_path) except Exception as exc:
_write_file(new_path, file_binary) logging.error(f'Could not write file: {exc}', exc_info=True)
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): def _get_counted_file_path(original_path):
...@@ -95,55 +76,46 @@ def _get_counted_file_path(original_path): ...@@ -95,55 +76,46 @@ def _get_counted_file_path(original_path):
return new_file_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. Fail-safe delete file operation. Deletes a file if it exists.
Errors are logged. No exception raised. Errors are logged. No exception raised.
:param file_path: Path of the file. Can be absolute or relative to the current directory. :param file_path: Path of the file. Can be absolute or relative to the current directory.
:type file_path: str
:return: None
''' '''
try: try:
os.unlink(file_path) Path(file_path).unlink()
except Exception as e: except Exception as exc:
logging.error('Could not delete file: {} {}'.format(sys.exc_info()[0].__name__, e)) logging.error(f'Could not delete file: {exc}', exc_info=True)
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. Fail-safe symlink operation. Symlinks a file if dest does not exist.
Errors are logged. No exception raised. Errors are logged. No exception raised.
:param src_path: src file :param src_path: src file
:type src_path: str
:param dst_path: link location :param dst_path: link location
:type dst_path: str
:return: None
''' '''
try: try:
create_dir_for_file(dst_path) create_dir_for_file(dst_path)
os.symlink(src_path, dst_path) Path(dst_path).symlink_to(src_path)
except FileExistsError as e: except FileExistsError as exc:
logging.debug('Could not create Link: File exists: {}'.format(e)) logging.debug(f'Could not create Link: File exists: {exc}')
except Exception as e: except Exception as exc:
logging.error('Could not create link: {} {}'.format(sys.exc_info()[0].__name__, e)) logging.error(f'Could not create link: {exc}', exc_info=True)
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 removes all problematic characters from a file name
cuts file names if they are too long cuts file names if they are too long
:param file_name: Original file name :param file_name: Original file name
:type file_name: str
:param max_size: maximum allowed file name length :param max_size: maximum allowed file name length
:type max_size: int
:default max_size: 200 :default max_size: 200
:param valid_characters: characters that shall be allowed in a file name :param valid_characters: characters that shall be allowed in a file name
:type valid_characters: str
:default valid_characters: 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_+. ' :default valid_characters: 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_+. '
:return: str
''' '''
allowed_charachters = set(valid_characters) allowed_charachters = set(valid_characters)
safe_name = filter(lambda x: x in allowed_charachters, file_name) safe_name = filter(lambda x: x in allowed_charachters, file_name)
...@@ -154,56 +126,53 @@ def get_safe_name(file_name, max_size=200, valid_characters='abcdefghijklmnopqrs ...@@ -154,56 +126,53 @@ def get_safe_name(file_name, max_size=200, valid_characters='abcdefghijklmnopqrs
return safe_name 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 Returns a list with the absolute paths of all files in the directory directory_path
:param directory_path: directory including files :param directory_path: directory including files
:type directory_path: str
:return: list
''' '''
result = [] result = []
try: try:
for file_path, _, files in os.walk(directory_path): for file_path, _, files in os.walk(directory_path):
for file_ in files: 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: except Exception as exc:
logging.error('Could not get files: {} {}'.format(sys.exc_info()[0].__name__, e)) logging.error(f'Could not get files: {exc}', exc_info=True)
return result 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. 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 :param directory_path: directory including sub-directories
:type directory_path: str
:return: list
''' '''
result = [] result = []
try: try:
dir_content = os.listdir(directory_path) path = Path(directory_path)
for item in dir_content: for item in path.iterdir():
dir_path = os.path.join(directory_path, item) if Path(item).is_dir():
if os.path.isdir(dir_path): result.append(str(item.resolve()))
result.append(dir_path) except Exception as exc:
except Exception as e: logging.error(f'Could not get directories: {exc}', exc_info=True)
logging.error('Could not get directories: {} {}'.format(sys.exc_info()[0].__name__, e))
return result 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 Returns absolute path of the directory including file
:param file_path: Path of the file :param file_path: Path of the file
:type: path-like object
:return: string .. deprecated::
You should use pathlib instead of this function.
''' '''
try: try:
return os.path.dirname(os.path.abspath(file_path)) return str(Path(file_path).resolve().parent)
except Exception as e: except Exception as exc:
logging.error('Could not get directory path: {} {}'.format(sys.exc_info()[0].__name__, e)) logging.error(f'Could not get directory path: {exc}', exc_info=True)
return '/' return '/'
...@@ -231,7 +200,7 @@ def _iterate_path_recursively(path: Path, include_symlinks: bool = True, include ...@@ -231,7 +200,7 @@ def _iterate_path_recursively(path: Path, include_symlinks: bool = True, include
for child_path in path.iterdir(): for child_path in path.iterdir():
yield from _iterate_path_recursively(child_path, include_symlinks, include_directories) yield from _iterate_path_recursively(child_path, include_symlinks, include_directories)
except PermissionError: except PermissionError:
logging.error('Permission Error: could not access path {path}'.format(path=path.absolute())) logging.error(f'Permission Error: could not access path {path.absolute()}')
except OSError: except OSError:
logging.warning('possible broken symlink: {path}'.format(path=path.absolute())) logging.warning(f'possible broken symlink: {path.absolute()}')
yield from [] yield from []
import os import io
from pathlib import Path
from typing import Type, Union
import bitmath 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:: 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): ...@@ -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 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. :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. :return: Returns a generator to iterate over all chunks, see above for usage.
''' '''
while True: while True:
...@@ -22,35 +24,36 @@ def read_in_chunks(file_object, chunk_size=1024): ...@@ -22,35 +24,36 @@ def read_in_chunks(file_object, chunk_size=1024):
yield data 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. 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. :param filename: Path of the file. Can be absolute or relative to the current directory.
:type filename: str
:return: Absolute path of the directory :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. 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. :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)) Path(file_path).resolve().parent.mkdir(parents=True, exist_ok=True)
os.makedirs(directory, exist_ok=True)
def human_readable_file_size(size_in_bytes): def human_readable_file_size(size_in_bytes: int) -> str:
''' '''
Returns a nicly human readable file size Returns a nicely human readable file size
:param size_in_bytes: Size in Bytes :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}') return bitmath.Byte(bytes=size_in_bytes).best_prefix().format('{value:.2f} {unit}')
import subprocess 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') return subprocess.check_output(['git', 'describe', '--always'], cwd=directory_name).strip().decode('utf-8')
from setuptools import setup, find_packages from setuptools import setup, find_packages
VERSION = '0.2.3' VERSION = '0.3.0'
setup( setup(
name='common_helper_files', name='common_helper_files',
......
...@@ -8,7 +8,7 @@ from common_helper_files import ( ...@@ -8,7 +8,7 @@ from common_helper_files import (
create_symlink, delete_file, get_safe_name, get_binary_from_file, get_dir_of_file, get_directory_for_filename, create_symlink, delete_file, get_safe_name, get_binary_from_file, get_dir_of_file, get_directory_for_filename,
get_dirs_in_dir, get_files_in_dir, get_string_list_from_file, safe_rglob, write_binary_to_file get_dirs_in_dir, get_files_in_dir, get_string_list_from_file, safe_rglob, write_binary_to_file
) )
from common_helper_files.fail_safe_file_operations import _get_counted_file_path, _rm_cr from common_helper_files.fail_safe_file_operations import _get_counted_file_path
TEST_DATA_DIR = Path(__file__).absolute().parent / 'data' TEST_DATA_DIR = Path(__file__).absolute().parent / 'data'
EMPTY_FOLDER = TEST_DATA_DIR / 'empty_folder' EMPTY_FOLDER = TEST_DATA_DIR / 'empty_folder'
...@@ -188,11 +188,3 @@ def test_safe_rglob_empty_dir(): ...@@ -188,11 +188,3 @@ def test_safe_rglob_empty_dir():
assert EMPTY_FOLDER.exists() assert EMPTY_FOLDER.exists()
result = safe_rglob(EMPTY_FOLDER) result = safe_rglob(EMPTY_FOLDER)
assert len(list(result)) == 0 assert len(list(result)) == 0
@pytest.mark.parametrize('input_data, expected', [
('abc', 'abc'),
('ab\r\nc', 'ab\nc'),
])
def test_rm_cr(input_data, expected):
assert _rm_cr(input_data) == expected
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