1
|
|
|
import os |
2
|
|
|
import tempfile |
3
|
|
|
from contextlib import contextmanager |
4
|
|
|
from typing import Dict, Union, TextIO |
5
|
|
|
|
6
|
|
|
from dirutility.error import InvalidAbsoluteDirectoryError |
7
|
|
|
from dirutility.open.clazz import TempDir, TarFile, ZipFile |
8
|
|
|
|
9
|
|
|
|
10
|
|
View Code Duplication |
@contextmanager |
|
|
|
|
11
|
|
|
def tempdir(absdirpath: str, prefix='', suffix='') -> TempDir: |
12
|
|
|
""" |
13
|
|
|
Create and return a temporary directory. This has the same |
14
|
|
|
behavior as mkdtemp but can be used as a context manager. For |
15
|
|
|
example: |
16
|
|
|
|
17
|
|
|
with tempdir('/home/dirutility/temp', prefix='prefix_', suffix='_suffix') as tmpdir: |
18
|
|
|
... |
19
|
|
|
|
20
|
|
|
Upon exiting the context, the directory and everything contained |
21
|
|
|
in it are removed. |
22
|
|
|
|
23
|
|
|
:param absdirpath: |
24
|
|
|
:param prefix: |
25
|
|
|
:param suffix: |
26
|
|
|
:return: |
27
|
|
|
:rtype: TempDir |
28
|
|
|
""" |
29
|
|
|
with tempfile.TemporaryDirectory(prefix=prefix, suffix=suffix, dir=absdirpath) as tmpdir_path: |
30
|
|
|
if os.path.isabs(absdirpath) and os.path.isdir(absdirpath): |
31
|
|
|
dir_obj = TempDir(tmpdir_path) |
32
|
|
|
yield dir_obj |
33
|
|
|
else: |
34
|
|
|
raise InvalidAbsoluteDirectoryError(absdirpath) |
35
|
|
|
|
36
|
|
|
|
37
|
|
View Code Duplication |
@contextmanager |
|
|
|
|
38
|
|
|
def tartempdir(absfilepath: str, |
39
|
|
|
tempdir: TempDir, |
40
|
|
|
mode: str = 'w:gz', |
41
|
|
|
temp: bool = True, |
42
|
|
|
force: bool = True) -> TarFile: |
43
|
|
|
""" |
44
|
|
|
Create and return a tarfile directory. This has the same |
45
|
|
|
behavior as mkdtemp then create tarfile from temp dir but can be used as a context manager. For |
46
|
|
|
example: |
47
|
|
|
|
48
|
|
|
with tardir('/home/dirutility/temp/bac.tar', 'haha.tar') as tardir: |
49
|
|
|
... |
50
|
|
|
|
51
|
|
|
Upon exiting the context, the directory and everything contained |
52
|
|
|
in it are removed. |
53
|
|
|
|
54
|
|
|
:param absfilepath: archive file create abs path |
55
|
|
|
:param tempdir: temporary directory object |
56
|
|
|
:param mode: tarfile mode |
57
|
|
|
:param temp: temporary file |
58
|
|
|
:param force: force create file |
59
|
|
|
:return: |
60
|
|
|
:rtype: TarFile |
61
|
|
|
""" |
62
|
|
|
file_create_by_me = False |
63
|
|
|
try: |
64
|
|
|
if force: |
65
|
|
|
yield _create_archive(absfilepath, tempdir, mode) |
66
|
|
|
file_create_by_me = True |
67
|
|
|
elif not os.path.exists(absfilepath): |
68
|
|
|
yield _create_archive(absfilepath, tempdir, mode) |
69
|
|
|
else: |
70
|
|
|
raise FileExistsError(absfilepath) |
71
|
|
|
finally: |
72
|
|
|
if temp and file_create_by_me: |
73
|
|
|
os.remove(absfilepath) |
74
|
|
|
else: |
75
|
|
|
pass |
76
|
|
|
|
77
|
|
|
|
78
|
|
|
def _create_archive(absfilepath, tempdir, mode) -> TarFile: |
79
|
|
|
""" |
80
|
|
|
|
81
|
|
|
:param absfilepath: |
82
|
|
|
:param tempdir: |
83
|
|
|
:param mode: |
84
|
|
|
:return: |
85
|
|
|
""" |
86
|
|
|
if mode.endswith('zip'): |
87
|
|
|
return ZipFile(absfilepath, tempdir, mode) |
88
|
|
|
else: |
89
|
|
|
return TarFile(absfilepath, tempdir, mode) |
90
|
|
|
|
91
|
|
|
|
92
|
|
View Code Duplication |
@contextmanager |
|
|
|
|
93
|
|
|
def tardir(absfilepath: str, |
94
|
|
|
mode: str = 'w:gz', |
95
|
|
|
temp: bool = True, |
96
|
|
|
force: bool = True, |
97
|
|
|
withdir: bool = False, |
98
|
|
|
**paths: Dict[str, Union[dict, str, TextIO, None]]) -> TarFile: |
99
|
|
|
""" |
100
|
|
|
Create and return a tarfile directory. This has the same |
101
|
|
|
behavior as mkdtemp then create tarfile from temp dir but can be used as a context manager. For |
102
|
|
|
example: |
103
|
|
|
|
104
|
|
|
with tardir('/home/dirutility/temp/bac.tar', 'haha.tar') as tardir: |
105
|
|
|
... |
106
|
|
|
|
107
|
|
|
Upon exiting the context, the directory and everything contained |
108
|
|
|
in it are removed. |
109
|
|
|
|
110
|
|
|
:param absfilepath: archive file create abs path |
111
|
|
|
:param mode: tarfile mode |
112
|
|
|
:param temp: temporary file |
113
|
|
|
:param force: force create file |
114
|
|
|
:param withdir: if or not with dir in archive file |
115
|
|
|
:param paths: directory paths objects |
116
|
|
|
:return: |
117
|
|
|
:rtype: TarFile |
118
|
|
|
""" |
119
|
|
|
absdirpath = absfilepath.rsplit('/', 1)[0] |
120
|
|
|
with tempdir(absdirpath) as temp_obj: |
121
|
|
|
temp_obj.create_structure(**paths) |
122
|
|
|
with tartempdir(absfilepath, temp_obj, mode=mode, temp=temp, force=force) as tar_obj: |
123
|
|
|
tar_obj.create_archive(withdir=withdir) |
124
|
|
|
yield tar_obj |
125
|
|
|
|