Coverage for autocrud/fastapi_generator.py: 52%

1"""FastAPI 自動生成模組""" 

2 

3from typing import Dict, Optional, Type 

4from fastapi import FastAPI, HTTPException, status 

5from pydantic import BaseModel, create_model 

6from .core import AutoCRUD 

7from .converter import ModelConverter 

8 

9 

10class FastAPIGenerator: 

11 """FastAPI 路由自動生成器""" 

12 

13 def __init__(self, crud: AutoCRUD): 

14 self.crud = crud 

15 self.converter = ModelConverter() 

16 

17 # 生成 Pydantic 模型用於請求/響應 

18 self.request_model = self._create_request_model() 

19 self.response_model = self._create_response_model() 

20 

21 def _create_request_model(self) -> Type[BaseModel]: 

22 """創建請求模型（不包含 ID）""" 

23 fields = self.converter.extract_fields(self.crud.model) 

24 

25 # 移除 ID 欄位（如果存在） 

26 fields.pop("id", None) 

27 

28 # 創建 Pydantic 模型 

29 return create_model( 

30 f"{self.crud.model.__name__}Request", 

31 **{name: (field_type, ...) for name, field_type in fields.items()}, 

32 ) 

33 

34 def _create_response_model(self) -> Type[BaseModel]: 

35 """創建響應模型（包含 ID）""" 

36 fields = self.converter.extract_fields(self.crud.model) 

37 

38 # 確保包含 ID 欄位 

39 fields["id"] = str 

40 

41 # 創建 Pydantic 模型 

42 return create_model( 

43 f"{self.crud.model.__name__}Response", 

44 **{name: (field_type, ...) for name, field_type in fields.items()}, 

45 ) 

46 

47 def create_routes(self, app: FastAPI, prefix: str = "") -> FastAPI: 

48 """在 FastAPI 應用中創建 CRUD 路由""" 

49 

50 resource_path = f"{prefix}/{self.crud.resource_name}" 

51 request_model = self.request_model 

52 response_model = self.response_model 

53 crud = self.crud 

54 

55 @app.post( 

56 f"{resource_path}", 

57 response_model=response_model, 

58 status_code=status.HTTP_201_CREATED, 

59 ) 

60 async def create_resource(item): 

61 """創建資源""" 

62 try: 

63 item_dict = item.model_dump() 

64 created_item = crud.create(item_dict) 

65 return created_item 

66 except Exception as e: 

67 raise HTTPException( 

68 status_code=status.HTTP_400_BAD_REQUEST, 

69 detail=f"創建失敗: {str(e)}", 

70 ) 

71 

72 # 設定類型提示 

73 create_resource.__annotations__["item"] = request_model 

74 

75 @app.get(f"{resource_path}/{{resource_id}}", response_model=response_model) 

76 async def get_resource(resource_id: str): 

77 """獲取單個資源""" 

78 item = crud.get(resource_id) 

79 if item is None: 

80 raise HTTPException( 

81 status_code=status.HTTP_404_NOT_FOUND, 

82 detail=f"資源不存在: {resource_id}", 

83 ) 

84 return item 

85 

86 @app.put(f"{resource_path}/{{resource_id}}", response_model=response_model) 

87 async def update_resource(resource_id: str, item): 

88 """更新資源""" 

89 if not crud.exists(resource_id): 

90 raise HTTPException( 

91 status_code=status.HTTP_404_NOT_FOUND, 

92 detail=f"資源不存在: {resource_id}", 

93 ) 

94 

95 try: 

96 item_dict = item.model_dump() 

97 updated_item = crud.update(resource_id, item_dict) 

98 return updated_item 

99 except Exception as e: 

100 raise HTTPException( 

101 status_code=status.HTTP_400_BAD_REQUEST, 

102 detail=f"更新失敗: {str(e)}", 

103 ) 

104 

105 # 設定類型提示 

106 update_resource.__annotations__["item"] = request_model 

107 

108 @app.delete( 

109 f"{resource_path}/{{resource_id}}", status_code=status.HTTP_204_NO_CONTENT 

110 ) 

111 async def delete_resource(resource_id: str): 

112 """刪除資源""" 

113 if not crud.exists(resource_id): 

114 raise HTTPException( 

115 status_code=status.HTTP_404_NOT_FOUND, 

116 detail=f"資源不存在: {resource_id}", 

117 ) 

118 

119 success = crud.delete(resource_id) 

120 if not success: 

121 raise HTTPException( 

122 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="刪除失敗" 

123 ) 

124 

125 @app.get(f"{resource_path}", response_model=Dict[str, response_model]) 

126 async def list_resources(): 

127 """列出所有資源""" 

128 return crud.list_all() 

129 

130 return app 

131 

132 def create_fastapi_app( 

133 self, 

134 title: Optional[str] = None, 

135 description: Optional[str] = None, 

136 version: str = "1.0.0", 

137 prefix: str = "/api/v1", 

138 ) -> FastAPI: 

139 """創建完整的 FastAPI 應用""" 

140 

141 if title is None: 

142 title = f"{self.crud.model.__name__} API" 

143 

144 if description is None: 

145 description = f"自動生成的 {self.crud.model.__name__} CRUD API" 

146 

147 app = FastAPI(title=title, description=description, version=version) 

148 

149 # 添加健康檢查端點 

150 @app.get("/health") 

151 async def health_check(): 

152 return {"status": "healthy", "service": title} 

153 

154 # 添加 CRUD 路由 

155 self.create_routes(app, prefix) 

156 

157 return app 

158 

159 

160# 為了向後兼容，在 AutoCRUD 類中添加 create_fastapi_app 方法 

161def create_fastapi_app_method(self, **kwargs) -> FastAPI: 

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

163 generator = FastAPIGenerator(self) 

164 return generator.create_fastapi_app(**kwargs) 

165 

166 

167# 使用範例 

168if __name__ == "__main__": 

169 from dataclasses import dataclass 

170 from .storage import MemoryStorage 

171 from .core import AutoCRUD 

172 

173 @dataclass 

174 class User: 

175 name: str 

176 email: str 

177 age: int 

178 

179 # 創建 CRUD 系統 

180 storage = MemoryStorage() 

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

182 

183 # 生成 FastAPI 應用 

184 generator = FastAPIGenerator(crud) 

185 app = generator.create_fastapi_app( 

186 title="用戶管理 API", description="自動生成的用戶 CRUD API" 

187 ) 

188 

189 print("FastAPI 應用已創建！") 

190 print("可用端點:") 

191 print("- POST /api/v1/users") 

192 print("- GET /api/v1/users/{id}") 

193 print("- PUT /api/v1/users/{id}") 

194 print("- DELETE /api/v1/users/{id}") 

195 print("- GET /api/v1/users")