Repository: sfneal/dirutility (language: python)

Issue:
NEW functionality of create directory
* Create temporary directory with an `with` syntax of opendir()
    - create and holding the directory if not exists
    - any operation in the directory
    - delete directory after leave `with`


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..580fea8
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,9 @@
+# ignore package
+**/*.egg-info/
+**/dist/
+
+# idea
+**/.idea/
+
+# python env
+**/*.pyenv
diff --git a/dirutility/error.py b/dirutility/error.py
new file mode 100644
index 0000000..ecdc6a8
--- /dev/null
+++ b/dirutility/error.py
@@ -0,0 +1,10 @@
+class InvalidFileNameError(Exception):
+    pass
+
+
+class InvalidAbsoluteDirectoryError(Exception):
+    pass
+
+
+class InvalidDirStructureError(Exception):
+    pass
diff --git a/dirutility/open/__init__.py b/dirutility/open/__init__.py
new file mode 100644
index 0000000..4159b3f
--- /dev/null
+++ b/dirutility/open/__init__.py
@@ -0,0 +1,3 @@
+__all__ = ['tempdir']
+
+from dirutility.open.open import tempdir
diff --git a/dirutility/open/clazz.py b/dirutility/open/clazz.py
new file mode 100644
index 0000000..83fc0c6
--- /dev/null
+++ b/dirutility/open/clazz.py
@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import abc
+import os
+import tarfile
+from typing import Union, Dict, TextIO, List
+from zipfile import PyZipFile
+
+from dirutility.error import InvalidFileNameError, InvalidDirStructureError
+
+
+class DirStructure:
+    def __init__(self, **paths: Dict[str, Union[dict, str, TextIO, None]]):
+        self._content = {}
+        for pathname, _c in paths.items():
+            if isinstance(_c, dict):
+                self._content[pathname] = DirStructure(**_c)
+            else:
+                self._content[pathname] = _c
+
+    @property
+    def content(self):
+        return self._content.copy()
+
+    @property
+    def dirsname(self) -> List[str]:
+        res = list(self._content.keys())
+        res.sort()
+        return res
+
+    def _is_filename_valid(self, filename):
+        if isinstance(filename, str) and filename.strip():
+            return True
+        else:
+            return False
+
+    def validate_filename(self, filename):
+        if not self._is_filename_valid(filename):
+            raise InvalidFileNameError(filename)
+
+    def create(self, prefix_dir: str):
+        for filename, _c in self._content.items():
+            self.validate_filename(filename)
+            path = '{}/{}'.format(prefix_dir, filename)
+            if _c is None:
+                fd = open(path, 'w')
+                fd.close()
+            if isinstance(_c, str):
+                with open(path, 'w') as fd:
+                    fd.write(_c)
+            elif isinstance(_c, dict):
+                os.mkdir(path)
+            elif isinstance(_c, DirStructure):
+                os.mkdir(path)
+                _c.create(path)
+            else:
+                with open(path, 'w') as fd:
+                    for _l in _c:
+                        fd.write(_l)
+
+
+class DirBase(metaclass=abc.ABCMeta):
+    def __init__(self, dirpath: str, structure: Union[DirStructure, None] = None):
+        self._dirpath = dirpath
+        self._structure = structure
+
+    @property
+    def dirname(self) -> str:
+        return self.dirpath.rsplit('/', 1)[1]
+
+    @property
+    def dirpath(self) -> str:
+        return self._dirpath
+
+    @property
+    def dirparent(self) -> str:
+        return self.dirpath.rsplit('/', 1)[0]
+
+    def dirs(self, relative=True) -> List[str]:
+        if relative:
+            return self._structure.dirsname
+        else:
+            return ['{}/{}'.format(self._dirpath, _name) for _name in self._structure.dirsname]
+
+    def create_structure(self, **paths: Dict[str, Union[dict, str, TextIO, None]]):
+        self._structure = DirStructure(**paths)
+        self._structure.create(self._dirpath)
+
+
+class TempDir(DirBase):
+    pass
+
+
+class TarFile(object):
+    def __init__(self, filepath: str, tempdir: TempDir, mode: str = 'w:gz'):
+
+        self._mode = mode.strip()
+        _ar = self._mode.split(':')
+        if len(_ar) > 1 and _ar[-1].strip():
+            self._compress_type = _ar[-1].strip()
+        else:
+            self._compress_type = None
+        self._open_mode = _ar[0]
+        self._filepath = filepath.strip()
+        self._filename = self._filepath.rsplit('/', 1)[-1]
+        self._filename_without_suffix = self._filename.rsplit('.', 1)[0]
+        self._validate_filename(self._filepath)
+        self._tempdir = tempdir
+
+    def _is_filename_valid(self, filename):
+        if filename.endswith(self._compress_type):
+            return True
+        else:
+            return False
+
+    def _validate_filename(self, filename):
+        if not self._is_filename_valid(filename):
+            raise InvalidFileNameError(filename)
+
+    @property
+    def filepath(self) -> str:
+        return self._filepath
+
+    def filename(self, with_suffix=True) -> str:
+        if with_suffix:
+            return self._filename
+        else:
+            return self._filename_without_suffix
+
+    @property
+    def tempdir(self) -> TempDir:
+        return self._tempdir
+
+    def create_archive(self, withdir: bool = False):
+        with tarfile.open(self._filepath, mode=self._mode) as tar:
+            for dirpath, dirname in zip(self._tempdir.dirs(relative=False), self._tempdir.dirs()):
+                tar.add(dirpath, arcname=dirname, recursive=True)
+
+
+class ZipFile(TarFile):
+    def __init__(self, filepath: str, tempdir: TempDir, mode: str = 'w:zip'):
+        super(ZipFile, self).__init__(filepath, tempdir, mode)
+
+    def create_archive(self, withdir: bool = False):
+        zip_file = PyZipFile(self._filepath, mode=self._open_mode)
+        for root, dirs, files in os.walk(self._tempdir.dirpath):
+            if files:
+                for file in files:
+                    _fp = os.path.join(root, file)
+                    if withdir:
+                        zip_file.write(_fp, arcname='{}/{}'.format(
+                            self.filename(with_suffix=False), _fp.replace(self._tempdir.dirpath, '')))
+                    else:
+                        zip_file.write(_fp, arcname=_fp.replace(self._tempdir.dirpath, ''))
+
+            else:
+                if withdir:
+                    zip_file.write(root, arcname='{}/{}'.format(
+                        self.filename(with_suffix=False), root.replace(self._tempdir.dirpath, '')))
+                else:
+                    zip_file.write(root, arcname=root.replace(self._tempdir.dirpath, ''))
diff --git a/dirutility/open/open.py b/dirutility/open/open.py
new file mode 100644
index 0000000..5e389d5
--- /dev/null
+++ b/dirutility/open/open.py
@@ -0,0 +1,118 @@
+import os
+import tempfile
+from contextlib import contextmanager
+from typing import Dict, Union, TextIO
+
+from dirutility.error import InvalidAbsoluteDirectoryError
+from dirutility.open.clazz import TempDir, TarFile, ZipFile
+
+
+@contextmanager
+def tempdir(absdirpath: str, prefix='', suffix='') -> TempDir:
+    """
+    Create and return a temporary directory.  This has the same
+    behavior as mkdtemp but can be used as a context manager.  For
+    example:
+
+        with tempdir('/home/dirutility/temp', prefix='prefix_', suffix='_suffix') as tmpdir:
+            ...
+
+    Upon exiting the context, the directory and everything contained
+    in it are removed.
+
+    :param absdirpath:
+    :param prefix:
+    :param suffix:
+    :return:
+    :rtype: TempDir
+    """
+    with tempfile.TemporaryDirectory(prefix=prefix, suffix=suffix, dir=absdirpath) as tmpdir_path:
+        if os.path.isabs(absdirpath) and os.path.isdir(absdirpath):
+            dir_obj = TempDir(tmpdir_path)
+            yield dir_obj
+        else:
+            raise InvalidAbsoluteDirectoryError(absdirpath)
+
+
+@contextmanager
+def tartempdir(
+        absfilepath: str, tempdir: TempDir, mode: str = 'w:gz', temp: bool = True, force: bool = True) -> TarFile:
+    """
+    Create and return a tarfile directory.  This has the same
+    behavior as mkdtemp then create tarfile from temp dir but can be used as a context manager.  For
+    example:
+
+        with tardir('/home/dirutility/temp/bac.tar', 'haha.tar') as tardir:
+            ...
+
+    Upon exiting the context, the directory and everything contained
+    in it are removed.
+
+    :param absfilepath: archive file create abs path
+    :param tempdir: temporary directory object
+    :param mode: tarfile mode
+    :param temp: temporary file
+    :param force: force create file
+    :return:
+    :rtype: TarFile
+    """
+    file_create_by_me = False
+    try:
+        if force:
+            yield _create_archive(absfilepath, tempdir, mode)
+            file_create_by_me = True
+        elif not os.path.exists(absfilepath):
+            yield _create_archive(absfilepath, tempdir, mode)
+        else:
+            raise FileExistsError(absfilepath)
+    finally:
+        if temp and file_create_by_me:
+            os.remove(absfilepath)
+        else:
+            pass
+
+
+def _create_archive(absfilepath, tempdir, mode) -> TarFile:
+    """
+
+    :param absfilepath:
+    :param tempdir:
+    :param mode:
+    :return:
+    """
+    if mode.endswith('zip'):
+        return ZipFile(absfilepath, tempdir, mode)
+    else:
+        return TarFile(absfilepath, tempdir, mode)
+
+
+@contextmanager
+def tardir(
+        absfilepath: str, mode: str = 'w:gz', temp: bool = True, force: bool = True, withdir: bool = False,
+        **paths: Dict[str, Union[dict, str, TextIO, None]]) -> TarFile:
+    """
+    Create and return a tarfile directory.  This has the same
+    behavior as mkdtemp then create tarfile from temp dir but can be used as a context manager.  For
+    example:
+
+        with tardir('/home/dirutility/temp/bac.tar', 'haha.tar') as tardir:
+            ...
+
+    Upon exiting the context, the directory and everything contained
+    in it are removed.
+
+    :param absfilepath: archive file create abs path
+    :param mode: tarfile mode
+    :param temp: temporary file
+    :param force: force create file
+    :param withdir: if or not with dir in archive file
+    :param paths: directory paths objects
+    :return:
+    :rtype: TarFile
+    """
+    absdirpath = absfilepath.rsplit('/', 1)[0]
+    with tempdir(absdirpath) as temp_obj:
+        temp_obj.create_structure(**paths)
+        with tartempdir(absfilepath, temp_obj, mode=mode, temp=temp, force=force) as tar_obj:
+            tar_obj.create_archive(withdir=withdir)
+            yield tar_obj


Test patch:
diff --git a/tests/test_archivefile.py b/tests/test_archivefile.py
new file mode 100644
index 0000000..c87cd79
--- /dev/null
+++ b/tests/test_archivefile.py
@@ -0,0 +1,54 @@
+import os
+import unittest
+from time import sleep
+
+from dirutility.open.open import tardir
+
+
+class TestArchive(unittest.TestCase):
+    def test_tardir(self):
+        tempfile = '/Users/Shared/kasdkfasdf.txt'
+        with open(tempfile, 'w') as wfd:
+            wfd.write(tempfile)
+        with open(tempfile, 'r') as rfd1, open(tempfile, 'r') as rfd2:
+            paths = {
+                'abc': 'abc abc abc',
+                'kbs': {
+                    'haha': {
+
+                    },
+                    'yoyo': {
+                        'haha': rfd1
+                    }
+                },
+                'abcd': {
+                    'kbs': rfd2
+                },
+            }
+            with tardir('/Users/Shared/haha.tar.gz', **paths) as tarfile:
+                self.assertTrue(os.path.exists(tarfile.filepath))
+            self.assertFalse(os.path.exists(tarfile.filepath))
+        with open(tempfile, 'r') as rfd1, open(tempfile, 'r') as rfd2:
+            paths = {
+                'abc': 'abc abc abc',
+                'kbs': {
+                    'haha': {
+
+                    },
+                    'yoyo': {
+                        'haha': rfd1
+                    }
+                },
+                'abcd': {
+                    'kbs': rfd2
+                },
+            }
+            with tardir('/Users/Shared/haha.zip', **paths, mode='w:zip', withdir=True) as tarfile:
+                self.assertTrue(os.path.exists(tarfile.filepath))
+                sleep(10)
+            self.assertFalse(os.path.exists(tarfile.filepath))
+        os.remove(tempfile)
+
+
+if __name__ == '__main__':
+    unittest.main()