Coverage for bookbuilderpy/temp.py: 97%

1"""A set of utilities interactions with the file system.""" 

2import os 

3from contextlib import AbstractContextManager 

4from shutil import rmtree 

5from tempfile import mkdtemp, mkstemp 

6 

7from bookbuilderpy.path import Path 

8from bookbuilderpy.strings import enforce_non_empty_str_without_ws 

9 

10 

11class TempDir(Path, AbstractContextManager): 

12 """ 

13 A scoped temporary directory to be used in a 'with' block. 

14 

15 The directory and everything in it will be deleted upon exiting the 

16 'with' block. 

17 """ 

18 

19 #: is the directory open? 

20 __is_open: bool 

21 

22 def __new__(cls, value): 

23 """ 

24 Construct the object. 

25 

26 :param value: the string value 

27 """ 

28 ret = super().__new__(cls, value) 

29 ret.enforce_dir() 

30 ret.__is_open = True 

31 return ret 

32 

33 @staticmethod 

34 def create(directory: str | None = None) -> "TempDir": 

35 """ 

36 Create the temporary directory. 

37 

38 :param directory: an optional root directory 

39 :raises TypeError: if `directory` is not `None` but also no `str` 

40 """ 

41 if directory is not None: 

42 root_dir = Path.path(directory) 

43 root_dir.enforce_dir() 

44 else: 

45 root_dir = None 

46 return TempDir(mkdtemp(dir=root_dir)) 

47 

48 def __enter__(self) -> "TempDir": 

49 """Nothing, just exists for `with`.""" 

50 if not self.__is_open: 

51 raise ValueError(f"Temporary directory '{self}' already closed.") 

52 return self 

53 

54 def __exit__(self, exception_type, exception_value, traceback) -> bool: 

55 """ 

56 Delete the temporary directory and everything in it. 

57 

58 :param exception_type: ignored 

59 :param exception_value: ignored 

60 :param traceback: ignored 

61 :returns: `True` to suppress an exception, `False` to rethrow it 

62 """ 

63 opn = self.__is_open 

64 self.__is_open = False 

65 if opn: 

66 rmtree(self, ignore_errors=True, onerror=None) 

67 return exception_type is None 

68 

69 

70class TempFile(Path, AbstractContextManager): 

71 """ 

72 A scoped temporary file to be used in a 'with' block. 

73 

74 This file will be deleted upon exiting the 'with' block. 

75 """ 

76 

77 #: is the directory open? 

78 __is_open: bool 

79 

80 def __new__(cls, value): 

81 """ 

82 Construct the object. 

83 

84 :param value: the string value 

85 """ 

86 ret = super().__new__(cls, value) 

87 ret.enforce_file() 

88 ret.__is_open = True 

89 return ret 

90 

91 @staticmethod 

92 def create(directory: str | None = None, 

93 prefix: str | None = None, 

94 suffix: str | None = None) -> "TempFile": 

95 """ 

96 Create a temporary file. 

97 

98 :param directory: a root directory or `TempDir` instance 

99 :param prefix: an optional prefix 

100 :param suffix: an optional suffix, e.g., `.txt` 

101 :raises TypeError: if any of the parameters does not fulfill the type 

102 contract 

103 """ 

104 if prefix is not None: 

105 prefix = enforce_non_empty_str_without_ws(prefix) 

106 

107 if suffix is not None: 

108 suffix = enforce_non_empty_str_without_ws(suffix) 

109 

110 if directory is not None: 

111 base_dir = Path.path(directory) 

112 base_dir.enforce_dir() 

113 else: 

114 base_dir = None 

115 

116 (handle, path) = mkstemp(suffix=suffix, prefix=prefix, dir=base_dir) 

117 os.close(handle) 

118 return TempFile(path) 

119 

120 def __enter__(self) -> "TempFile": 

121 """Nothing, just exists for `with`.""" 

122 if not self.__is_open: 

123 raise ValueError(f"Temporary file '{self}' already deleted.") 

124 return self 

125 

126 def __exit__(self, exception_type, exception_value, traceback) -> bool: 

127 """ 

128 Delete the temporary file. 

129 

130 :param exception_type: ignored 

131 :param exception_value: ignored 

132 :param traceback: ignored 

133 :returns: `True` to suppress an exception, `False` to rethrow it 

134 """ 

135 opn = self.__is_open 

136 self.__is_open = False 

137 if opn: 

138 os.remove(self) 

139 return exception_type is None