Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_shopping_lists.py: 65%

136 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:48 +0000

1from collections.abc import Callable 1a

2from functools import cached_property 1a

3 

4from fastapi import APIRouter, Depends, HTTPException, Query, status 1a

5from pydantic import UUID4 1a

6 

7from mealie.routes._base.base_controllers import BaseCrudController 1a

8from mealie.routes._base.controller import controller 1a

9from mealie.routes._base.mixins import HttpRepo 1a

10from mealie.schema.household.group_shopping_list import ( 1a

11 ShoppingListAddRecipeParams, 

12 ShoppingListAddRecipeParamsBulk, 

13 ShoppingListCreate, 

14 ShoppingListItemCreate, 

15 ShoppingListItemOut, 

16 ShoppingListItemPagination, 

17 ShoppingListItemsCollectionOut, 

18 ShoppingListItemUpdate, 

19 ShoppingListItemUpdateBulk, 

20 ShoppingListMultiPurposeLabelUpdate, 

21 ShoppingListOut, 

22 ShoppingListPagination, 

23 ShoppingListRemoveRecipeParams, 

24 ShoppingListSave, 

25 ShoppingListSummary, 

26 ShoppingListUpdate, 

27) 

28from mealie.schema.response.pagination import PaginationQuery 1a

29from mealie.schema.response.responses import SuccessResponse 1a

30from mealie.services.event_bus_service.event_types import ( 1a

31 EventOperation, 

32 EventShoppingListData, 

33 EventShoppingListItemBulkData, 

34 EventTypes, 

35) 

36from mealie.services.household_services.shopping_lists import ShoppingListService 1a

37 

38item_router = APIRouter(prefix="/households/shopping/items", tags=["Households: Shopping List Items"]) 1a

39 

40 

41def publish_list_item_events(publisher: Callable, items_collection: ShoppingListItemsCollectionOut) -> None: 1a

42 items_by_list_id: dict[UUID4, list[ShoppingListItemOut]] 

43 if items_collection.created_items: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true1efgb

44 items_by_list_id = {} 

45 for item in items_collection.created_items: 

46 items_by_list_id.setdefault(item.shopping_list_id, []).append(item) 

47 

48 for shopping_list_id, items in items_by_list_id.items(): 

49 publisher( 

50 EventTypes.shopping_list_updated, 

51 document_data=EventShoppingListItemBulkData( 

52 operation=EventOperation.create, 

53 shopping_list_id=shopping_list_id, 

54 shopping_list_item_ids=[item.id for item in items], 

55 ), 

56 # since these are all the same shopping list, they share a group_id and household_id 

57 group_id=items[0].group_id, 

58 household_id=items[0].household_id, 

59 ) 

60 

61 if items_collection.updated_items: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true1efgb

62 items_by_list_id = {} 

63 for item in items_collection.updated_items: 

64 items_by_list_id.setdefault(item.shopping_list_id, []).append(item) 

65 

66 for shopping_list_id, items in items_by_list_id.items(): 

67 publisher( 

68 EventTypes.shopping_list_updated, 

69 document_data=EventShoppingListItemBulkData( 

70 operation=EventOperation.update, 

71 shopping_list_id=shopping_list_id, 

72 shopping_list_item_ids=[item.id for item in items], 

73 ), 

74 # since these are all the same shopping list, they share a group_id and household_id 

75 group_id=items[0].group_id, 

76 household_id=items[0].household_id, 

77 ) 

78 

79 if items_collection.deleted_items: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true1efgb

80 items_by_list_id = {} 

81 for item in items_collection.deleted_items: 

82 items_by_list_id.setdefault(item.shopping_list_id, []).append(item) 

83 

84 for shopping_list_id, items in items_by_list_id.items(): 

85 publisher( 

86 EventTypes.shopping_list_updated, 

87 document_data=EventShoppingListItemBulkData( 

88 operation=EventOperation.delete, 

89 shopping_list_id=shopping_list_id, 

90 shopping_list_item_ids=[item.id for item in items], 

91 ), 

92 # since these are all the same shopping list, they share a group_id and household_id 

93 group_id=items[0].group_id, 

94 household_id=items[0].household_id, 

95 ) 

96 

97 

98@controller(item_router) 1a

99class ShoppingListItemController(BaseCrudController): 1a

100 @cached_property 1a

101 def service(self): 1a

102 return ShoppingListService(self.repos) 1efgb

103 

104 @cached_property 1a

105 def repo(self): 1a

106 return self.repos.group_shopping_list_item 

107 

108 @cached_property 1a

109 def mixins(self): 1a

110 return HttpRepo[ShoppingListItemCreate, ShoppingListItemOut, ShoppingListItemCreate]( 

111 self.repo, 

112 self.logger, 

113 ) 

114 

115 @item_router.get("", response_model=ShoppingListItemPagination) 1a

116 def get_all(self, q: PaginationQuery = Depends()): 1a

117 response = self.repo.page_all(pagination=q, override=ShoppingListItemOut) 

118 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

119 return response 

120 

121 @item_router.post("/create-bulk", response_model=ShoppingListItemsCollectionOut, status_code=201) 1a

122 def create_many(self, data: list[ShoppingListItemCreate]): 1a

123 items = self.service.bulk_create_items(data) 1fb

124 publish_list_item_events(self.publish_event, items) 1fb

125 return items 1fb

126 

127 @item_router.post("", response_model=ShoppingListItemsCollectionOut, status_code=201) 1a

128 def create_one(self, data: ShoppingListItemCreate): 1a

129 return self.create_many([data]) 

130 

131 @item_router.get("/{item_id}", response_model=ShoppingListItemOut) 1a

132 def get_one(self, item_id: UUID4): 1a

133 return self.mixins.get_one(item_id) 

134 

135 @item_router.put("", response_model=ShoppingListItemsCollectionOut) 1a

136 def update_many(self, data: list[ShoppingListItemUpdateBulk]): 1a

137 items = self.service.bulk_update_items(data) 1g

138 publish_list_item_events(self.publish_event, items) 1g

139 return items 1g

140 

141 @item_router.put("/{item_id}", response_model=ShoppingListItemsCollectionOut) 1a

142 def update_one(self, item_id: UUID4, data: ShoppingListItemUpdate): 1a

143 return self.update_many([data.cast(ShoppingListItemUpdateBulk, id=item_id)]) 

144 

145 @item_router.delete("", response_model=SuccessResponse) 1a

146 def delete_many(self, ids: list[UUID4] = Query(None)): 1a

147 items = self.service.bulk_delete_items(ids) 1e

148 publish_list_item_events(self.publish_event, items) 1e

149 return SuccessResponse.respond() 1e

150 

151 @item_router.delete("/{item_id}", response_model=SuccessResponse) 1a

152 def delete_one(self, item_id: UUID4): 1a

153 return self.delete_many([item_id]) 

154 

155 

156router = APIRouter(prefix="/households/shopping/lists", tags=["Households: Shopping Lists"]) 1a

157 

158 

159@controller(router) 1a

160class ShoppingListController(BaseCrudController): 1a

161 @cached_property 1a

162 def service(self): 1a

163 return ShoppingListService(self.repos) 1chijdklmnopqrb

164 

165 @cached_property 1a

166 def repo(self): 1a

167 return self.repos.group_shopping_lists 1db

168 

169 # ======================================================================= 

170 # CRUD Operations 

171 

172 @cached_property 1a

173 def mixins(self) -> HttpRepo[ShoppingListCreate, ShoppingListOut, ShoppingListSave]: 1a

174 return HttpRepo(self.repo, self.logger, self.registered_exceptions, self.t("generic.server-error")) 1db

175 

176 @router.get("", response_model=ShoppingListPagination) 1a

177 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a

178 response = self.repo.page_all( 

179 pagination=q, 

180 override=ShoppingListSummary, 

181 ) 

182 

183 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

184 return response 

185 

186 @router.post("", response_model=ShoppingListOut, status_code=201) 1a

187 def create_one(self, data: ShoppingListCreate): 1a

188 shopping_list = self.service.create_one_list(data, self.user.id) 1chijdklmnopqrb

189 if shopping_list: 189 ↛ 198line 189 didn't jump to line 198 because the condition on line 189 was always true1chijdklmnopqrb

190 self.publish_event( 1chijdklmnopqrb

191 event_type=EventTypes.shopping_list_created, 

192 document_data=EventShoppingListData(operation=EventOperation.create, shopping_list_id=shopping_list.id), 

193 group_id=shopping_list.group_id, 

194 household_id=shopping_list.household_id, 

195 message=self.t("notifications.generic-created", name=shopping_list.name), 

196 ) 

197 

198 return shopping_list 1chijdklmnopqrb

199 

200 @router.get("/{item_id}", response_model=ShoppingListOut) 1a

201 def get_one(self, item_id: UUID4): 1a

202 return self.mixins.get_one(item_id) 1d

203 

204 @router.put("/{item_id}", response_model=ShoppingListOut) 1a

205 def update_one(self, item_id: UUID4, data: ShoppingListUpdate): 1a

206 shopping_list = self.mixins.update_one(data, item_id) 

207 self.publish_event( 

208 event_type=EventTypes.shopping_list_updated, 

209 document_data=EventShoppingListData(operation=EventOperation.update, shopping_list_id=shopping_list.id), 

210 group_id=shopping_list.group_id, 

211 household_id=shopping_list.household_id, 

212 message=self.t("notifications.generic-updated", name=shopping_list.name), 

213 ) 

214 

215 return shopping_list 

216 

217 @router.delete("/{item_id}", response_model=ShoppingListOut) 1a

218 def delete_one(self, item_id: UUID4): 1a

219 shopping_list = self.mixins.delete_one(item_id) # type: ignore 

220 if shopping_list: 220 ↛ 229line 220 didn't jump to line 229 because the condition on line 220 was always true

221 self.publish_event( 

222 event_type=EventTypes.shopping_list_deleted, 

223 document_data=EventShoppingListData(operation=EventOperation.delete, shopping_list_id=shopping_list.id), 

224 group_id=shopping_list.group_id, 

225 household_id=shopping_list.household_id, 

226 message=self.t("notifications.generic-deleted", name=shopping_list.name), 

227 ) 

228 

229 return shopping_list 

230 

231 # ======================================================================= 

232 # Other Operations 

233 

234 @router.put("/{item_id}/label-settings", response_model=ShoppingListOut) 1a

235 def update_label_settings(self, item_id: UUID4, data: list[ShoppingListMultiPurposeLabelUpdate]): 1a

236 for setting in data: 

237 if setting.shopping_list_id != item_id: 

238 raise HTTPException( 

239 status_code=status.HTTP_400_BAD_REQUEST, 

240 detail=f"object {setting.id} has an invalid shopping list id", 

241 ) 

242 

243 self.repos.shopping_list_multi_purpose_labels.update_many(data) 

244 updated_list = self.get_one(item_id) 

245 

246 self.publish_event( 

247 event_type=EventTypes.shopping_list_updated, 

248 document_data=EventShoppingListData(operation=EventOperation.update, shopping_list_id=updated_list.id), 

249 group_id=updated_list.group_id, 

250 household_id=updated_list.household_id, 

251 message=self.t("notifications.generic-updated", name=updated_list.name), 

252 ) 

253 

254 return updated_list 

255 

256 @router.post("/{item_id}/recipe", response_model=ShoppingListOut) 1a

257 def add_recipe_ingredients_to_list(self, item_id: UUID4, data: list[ShoppingListAddRecipeParamsBulk]): 1a

258 shopping_list, items = self.service.add_recipe_ingredients_to_list(item_id, data) 1c

259 

260 publish_list_item_events(self.publish_event, items) 

261 return shopping_list 

262 

263 @router.post("/{item_id}/recipe/{recipe_id}", response_model=ShoppingListOut, deprecated=True) 1a

264 def add_single_recipe_ingredients_to_list( 1a

265 self, item_id: UUID4, recipe_id: UUID4, data: ShoppingListAddRecipeParams | None = None 

266 ): 

267 # Compatibility function for old API 

268 # TODO: remove this function in the future 

269 

270 data = data or ShoppingListAddRecipeParams(recipe_increment_quantity=1) 1c

271 bulk_data = [data.cast(ShoppingListAddRecipeParamsBulk, recipe_id=recipe_id)] 1c

272 return self.add_recipe_ingredients_to_list(item_id, bulk_data) 1c

273 

274 @router.post("/{item_id}/recipe/{recipe_id}/delete", response_model=ShoppingListOut) 1a

275 def remove_recipe_ingredients_from_list( 1a

276 self, item_id: UUID4, recipe_id: UUID4, data: ShoppingListRemoveRecipeParams | None = None 

277 ): 

278 shopping_list, items = self.service.remove_recipe_ingredients_from_list( 

279 item_id, recipe_id, data.recipe_decrement_quantity if data else 1 

280 ) 

281 

282 publish_list_item_events(self.publish_event, items) 

283 return shopping_list