Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_shopping_lists.py: 67%
136 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 17:29 +0000
1from collections.abc import Callable 1a
2from functools import cached_property 1a
4from fastapi import APIRouter, Depends, HTTPException, Query, status 1a
5from pydantic import UUID4 1a
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
38item_router = APIRouter(prefix="/households/shopping/items", tags=["Households: Shopping List Items"]) 1a
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 true
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)
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 )
61 if items_collection.updated_items: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true
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)
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 )
79 if items_collection.deleted_items: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
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)
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 )
98@controller(item_router) 1a
99class ShoppingListItemController(BaseCrudController): 1a
100 @cached_property 1a
101 def service(self): 1a
102 return ShoppingListService(self.repos)
104 @cached_property 1a
105 def repo(self): 1a
106 return self.repos.group_shopping_list_item
108 @cached_property 1a
109 def mixins(self): 1a
110 return HttpRepo[ShoppingListItemCreate, ShoppingListItemOut, ShoppingListItemCreate](
111 self.repo,
112 self.logger,
113 )
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
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)
124 publish_list_item_events(self.publish_event, items)
125 return items
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])
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)
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)
138 publish_list_item_events(self.publish_event, items)
139 return items
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)])
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)
148 publish_list_item_events(self.publish_event, items)
149 return SuccessResponse.respond()
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])
156router = APIRouter(prefix="/households/shopping/lists", tags=["Households: Shopping Lists"]) 1a
159@controller(router) 1a
160class ShoppingListController(BaseCrudController): 1a
161 @cached_property 1a
162 def service(self): 1a
163 return ShoppingListService(self.repos) 1ijklmenopcgDqrsthuvfdwxyzABb
165 @cached_property 1a
166 def repo(self): 1a
167 return self.repos.group_shopping_lists 1Cechfdb
169 # =======================================================================
170 # CRUD Operations
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")) 1echfdb
176 @router.get("", response_model=ShoppingListPagination) 1a
177 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a
178 response = self.repo.page_all( 1Cb
179 pagination=q,
180 override=ShoppingListSummary,
181 )
183 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1Cb
184 return response 1Cb
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) 1ijklmenopcgDqrsthuvfdwxyzABb
189 if shopping_list: 189 ↛ 198line 189 didn't jump to line 198 because the condition on line 189 was always true1ijklmenopcgqrsthuvfdwxyzABb
190 self.publish_event( 1ijklmenopcgqrsthuvfdwxyzABb
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 )
198 return shopping_list 1ijklmenopcgqrsthuvfdwxyzABb
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) 1cdb
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) 1chb
207 self.publish_event( 1c
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 )
215 return shopping_list 1c
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 1ecfdb
220 if shopping_list: 220 ↛ 229line 220 didn't jump to line 229 because the condition on line 220 was always true1ecfdb
221 self.publish_event( 1ecfdb
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 )
229 return shopping_list 1ecfdb
231 # =======================================================================
232 # Other Operations
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 )
243 self.repos.shopping_list_multi_purpose_labels.update_many(data)
244 updated_list = self.get_one(item_id)
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 )
254 return updated_list
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) 1gb
260 publish_list_item_events(self.publish_event, items)
261 return shopping_list
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
270 data = data or ShoppingListAddRecipeParams(recipe_increment_quantity=1) 1gb
271 bulk_data = [data.cast(ShoppingListAddRecipeParamsBulk, recipe_id=recipe_id)] 1gb
272 return self.add_recipe_ingredients_to_list(item_id, bulk_data) 1gb
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 )
282 publish_list_item_events(self.publish_event, items)
283 return shopping_list