Coverage for autocrud/core.py: 71%

1"""AutoCRUD 核心模組""" 

2 

3import uuid 

4from typing import Any, Dict, Type, Optional 

5from .converter import ModelConverter 

6from .storage import Storage 

7 

8 

9class AutoCRUD: 

10 """自動 CRUD 系統核心類""" 

11 

12 def __init__( 

13 self, 

14 model: Type, 

15 storage: Storage, 

16 resource_name: str, 

17 id_generator: Optional[callable] = None, 

18 ): 

19 self.model = model 

20 self.storage = storage 

21 self.resource_name = resource_name 

22 self.converter = ModelConverter() 

23 self.id_generator = id_generator or (lambda: str(uuid.uuid4())) 

24 

25 # 驗證模型類型 

26 self.model_type = self.converter.detect_model_type(model) 

27 self.model_fields = self.converter.extract_fields(model) 

28 

29 def _make_key(self, resource_id: str) -> str: 

30 """生成存儲鍵""" 

31 return f"{self.resource_name}:{resource_id}" 

32 

33 def create(self, data: Dict[str, Any]) -> Dict[str, Any]: 

34 """創建資源""" 

35 # 生成 ID 

36 resource_id = self.id_generator() 

37 

38 # 創建模型實例 

39 instance = self.converter.from_dict(self.model, data) 

40 

41 # 轉換為字典並添加 ID 

42 instance_dict = self.converter.to_dict(instance) 

43 instance_dict["id"] = resource_id 

44 

45 # 存儲 

46 key = self._make_key(resource_id) 

47 self.storage.set(key, instance_dict) 

48 

49 return instance_dict 

50 

51 def get(self, resource_id: str) -> Optional[Dict[str, Any]]: 

52 """獲取資源""" 

53 key = self._make_key(resource_id) 

54 return self.storage.get(key) 

55 

56 def update( 

57 self, resource_id: str, data: Dict[str, Any] 

58 ) -> Optional[Dict[str, Any]]: 

59 """更新資源""" 

60 key = self._make_key(resource_id) 

61 

62 # 檢查資源是否存在 

63 if not self.storage.exists(key): 

64 return None 

65 

66 # 創建模型實例驗證數據 

67 instance = self.converter.from_dict(self.model, data) 

68 

69 # 轉換為字典並保持 ID 

70 instance_dict = self.converter.to_dict(instance) 

71 instance_dict["id"] = resource_id 

72 

73 # 更新存儲 

74 self.storage.set(key, instance_dict) 

75 

76 return instance_dict 

77 

78 def delete(self, resource_id: str) -> bool: 

79 """刪除資源""" 

80 key = self._make_key(resource_id) 

81 return self.storage.delete(key) 

82 

83 def exists(self, resource_id: str) -> bool: 

84 """檢查資源是否存在""" 

85 key = self._make_key(resource_id) 

86 return self.storage.exists(key) 

87 

88 def list_all(self) -> Dict[str, Dict[str, Any]]: 

89 """列出所有資源""" 

90 prefix = f"{self.resource_name}:" 

91 all_keys = self.storage.list_keys() 

92 

93 result = {} 

94 for key in all_keys: 

95 if key.startswith(prefix): 

96 resource_id = key[len(prefix) :] 

97 data = self.storage.get(key) 

98 if data: 

99 result[resource_id] = data 

100 

101 return result 

102 

103 def create_fastapi_app(self, **kwargs): 

104 """創建 FastAPI 應用的便利方法""" 

105 from .fastapi_generator import FastAPIGenerator 

106 

107 generator = FastAPIGenerator(self) 

108 return generator.create_fastapi_app(**kwargs) 

109 

110 

111# 使用範例 

112if __name__ == "__main__": 

113 from dataclasses import dataclass 

114 from .storage import MemoryStorage 

115 

116 @dataclass 

117 class User: 

118 name: str 

119 email: str 

120 age: int 

121 

122 # 創建 AutoCRUD 實例 

123 storage = MemoryStorage() 

124 crud = AutoCRUD(model=User, storage=storage, resource_name="users") 

125 

126 # 測試創建 

127 user_data = {"name": "Alice", "email": "alice@example.com", "age": 30} 

128 created_user = crud.create(user_data) 

129 print(f"創建用戶: {created_user}") 

130 

131 # 測試獲取 

132 user_id = created_user["id"] 

133 retrieved_user = crud.get(user_id) 

134 print(f"獲取用戶: {retrieved_user}") 

135 

136 # 測試更新 

137 updated_data = { 

138 "name": "Alice Smith", 

139 "email": "alice.smith@example.com", 

140 "age": 31, 

141 } 

142 updated_user = crud.update(user_id, updated_data) 

143 print(f"更新用戶: {updated_user}") 

144 

145 # 測試列出所有 

146 all_users = crud.list_all() 

147 print(f"所有用戶: {all_users}") 

148 

149 # 測試刪除 

150 deleted = crud.delete(user_id) 

151 print(f"刪除成功: {deleted}") 

152 print(f"刪除後存在: {crud.exists(user_id)}")