Coverage for autocrud/storage.py: 64%

1"""存儲抽象層""" 

2 

3import os 

4from abc import ABC, abstractmethod 

5from typing import Any, Dict, List, Optional 

6from .serializer import Serializer, SerializerFactory 

7 

8 

9class Storage(ABC): 

10 """存儲抽象基類""" 

11 

12 def __init__(self, serializer: Optional[Serializer] = None): 

13 if serializer is None: 

14 serializer = SerializerFactory.create("json") 

15 self.serializer = serializer 

16 

17 @abstractmethod 

18 def get(self, key: str) -> Optional[Any]: 

19 """獲取數據""" 

20 pass 

21 

22 @abstractmethod 

23 def set(self, key: str, value: Any) -> None: 

24 """存儲數據""" 

25 pass 

26 

27 @abstractmethod 

28 def delete(self, key: str) -> bool: 

29 """刪除數據""" 

30 pass 

31 

32 @abstractmethod 

33 def exists(self, key: str) -> bool: 

34 """檢查鍵是否存在""" 

35 pass 

36 

37 @abstractmethod 

38 def list_keys(self) -> List[str]: 

39 """列出所有鍵""" 

40 pass 

41 

42 

43class MemoryStorage(Storage): 

44 """純內存存儲實現（重啟後數據消失，適合測試和演示）""" 

45 

46 def __init__(self, serializer: Optional[Serializer] = None): 

47 super().__init__(serializer) 

48 self._data: Dict[str, bytes] = {} 

49 

50 def get(self, key: str) -> Optional[Any]: 

51 if key not in self._data: 

52 return None 

53 

54 serialized_data = self._data[key] 

55 return self.serializer.deserialize(serialized_data) 

56 

57 def set(self, key: str, value: Any) -> None: 

58 serialized_data = self.serializer.serialize(value) 

59 self._data[key] = serialized_data 

60 

61 def delete(self, key: str) -> bool: 

62 if key in self._data: 

63 del self._data[key] 

64 return True 

65 return False 

66 

67 def exists(self, key: str) -> bool: 

68 return key in self._data 

69 

70 def list_keys(self) -> List[str]: 

71 return list(self._data.keys()) 

72 

73 def clear(self) -> None: 

74 """清空所有數據""" 

75 self._data.clear() 

76 

77 def size(self) -> int: 

78 """獲取存儲的條目數量""" 

79 return len(self._data) 

80 

81 

82class DiskStorage(Storage): 

83 """磁碟文件存儲實現（真正的持久化存儲）""" 

84 

85 def __init__(self, base_dir: str, serializer: Optional[Serializer] = None): 

86 super().__init__(serializer) 

87 self.base_dir = base_dir 

88 

89 # 確保基礎目錄存在 

90 os.makedirs(self.base_dir, exist_ok=True) 

91 

92 def _get_file_path(self, key: str) -> str: 

93 """獲取鍵對應的文件路徑""" 

94 # 將鍵轉換為安全的文件名 

95 safe_key = key.replace(":", "_").replace("/", "_") 

96 return os.path.join(self.base_dir, f"{safe_key}.data") 

97 

98 def get(self, key: str) -> Optional[Any]: 

99 file_path = self._get_file_path(key) 

100 if not os.path.exists(file_path): 

101 return None 

102 

103 try: 

104 with open(file_path, "rb") as f: 

105 serialized_data = f.read() 

106 return self.serializer.deserialize(serialized_data) 

107 except Exception as e: 

108 print(f"讀取文件失敗 {file_path}: {e}") 

109 return None 

110 

111 def set(self, key: str, value: Any) -> None: 

112 file_path = self._get_file_path(key) 

113 

114 try: 

115 serialized_data = self.serializer.serialize(value) 

116 with open(file_path, "wb") as f: 

117 f.write(serialized_data) 

118 except Exception as e: 

119 print(f"寫入文件失敗 {file_path}: {e}") 

120 raise 

121 

122 def delete(self, key: str) -> bool: 

123 file_path = self._get_file_path(key) 

124 if os.path.exists(file_path): 

125 try: 

126 os.remove(file_path) 

127 return True 

128 except Exception as e: 

129 print(f"刪除文件失敗 {file_path}: {e}") 

130 return False 

131 return False 

132 

133 def exists(self, key: str) -> bool: 

134 file_path = self._get_file_path(key) 

135 return os.path.exists(file_path) 

136 

137 def list_keys(self) -> List[str]: 

138 """列出所有鍵""" 

139 keys = [] 

140 try: 

141 for filename in os.listdir(self.base_dir): 

142 if filename.endswith(".data"): 

143 # 將文件名轉換回鍵名 

144 key = filename[:-5].replace("_", ":") # 移除 .data 後綴 

145 keys.append(key) 

146 except Exception as e: 

147 print(f"列出文件失敗: {e}") 

148 

149 return keys 

150 

151 def clear(self) -> None: 

152 """清空所有數據""" 

153 try: 

154 for filename in os.listdir(self.base_dir): 

155 if filename.endswith(".data"): 

156 file_path = os.path.join(self.base_dir, filename) 

157 os.remove(file_path) 

158 except Exception as e: 

159 print(f"清空目錄失敗: {e}") 

160 

161 def size(self) -> int: 

162 """獲取存儲的條目數量""" 

163 return len(self.list_keys()) 

164 

165 

166# 使用範例 

167if __name__ == "__main__": 

168 # 測試純內存存儲 

169 print("=== 測試純內存存儲 ===") 

170 memory_storage = MemoryStorage() 

171 

172 # 測試數據 

173 test_data = {"name": "Alice", "age": 30} 

174 

175 # 存儲數據 

176 memory_storage.set("user:1", test_data) 

177 print(f"內存存儲成功: {memory_storage.exists('user:1')}") 

178 

179 # 獲取數據 

180 retrieved_data = memory_storage.get("user:1") 

181 print(f"內存獲取數據: {retrieved_data}") 

182 

183 print(f"內存存儲大小: {memory_storage.size()}") 

184 

185 # 測試磁碟存儲 

186 print("\n=== 測試磁碟存儲 ===") 

187 import tempfile 

188 

189 temp_dir = tempfile.mkdtemp() 

190 disk_storage = DiskStorage(temp_dir) 

191 

192 # 存儲數據 

193 disk_storage.set("user:1", test_data) 

194 print(f"磁碟存儲成功: {disk_storage.exists('user:1')}") 

195 

196 # 獲取數據 

197 retrieved_data = disk_storage.get("user:1") 

198 print(f"磁碟獲取數據: {retrieved_data}") 

199 

200 print(f"磁碟存儲大小: {disk_storage.size()}") 

201 print(f"磁碟所有鍵: {disk_storage.list_keys()}") 

202 

203 # 清理 

204 disk_storage.clear() 

205 os.rmdir(temp_dir)