Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_cookbooks.py: 83%
80 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1from collections import defaultdict 1a
2from functools import cached_property 1a
3from uuid import UUID 1a
5from fastapi import APIRouter, Depends, HTTPException 1a
6from pydantic import UUID4 1a
8from mealie.core.exceptions import mealie_registered_exceptions 1a
9from mealie.repos.all_repositories import get_repositories 1a
10from mealie.routes._base import BaseCrudController, controller 1a
11from mealie.routes._base.mixins import HttpRepo 1a
12from mealie.routes._base.routers import MealieCrudRoute 1a
13from mealie.schema import mapper 1a
14from mealie.schema.cookbook import CreateCookBook, ReadCookBook, SaveCookBook, UpdateCookBook 1a
15from mealie.schema.cookbook.cookbook import CookBookPagination 1a
16from mealie.schema.response.pagination import PaginationQuery 1a
17from mealie.services.event_bus_service.event_types import ( 1a
18 EventCookbookBulkData,
19 EventCookbookData,
20 EventOperation,
21 EventTypes,
22)
24router = APIRouter(prefix="/households/cookbooks", tags=["Households: Cookbooks"], route_class=MealieCrudRoute) 1a
27@controller(router) 1a
28class GroupCookbookController(BaseCrudController): 1a
29 @cached_property 1a
30 def cookbooks(self): 1a
31 return self.repos.cookbooks 1pdehijcfklmgnb
33 @cached_property 1a
34 def group_cookbooks(self): 1a
35 return get_repositories(self.session, group_id=self.group_id, household_id=None).cookbooks 1dehijcfklmognb
37 def registered_exceptions(self, ex: type[Exception]) -> str: 1a
38 registered = {
39 **mealie_registered_exceptions(self.translator),
40 }
41 return registered.get(ex, self.t("generic.server-error"))
43 @cached_property 1a
44 def mixins(self): 1a
45 return HttpRepo[CreateCookBook, ReadCookBook, UpdateCookBook]( 1pdehijcfklmgnb
46 self.cookbooks,
47 self.logger,
48 self.registered_exceptions,
49 )
51 @router.get("", response_model=CookBookPagination) 1a
52 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a
53 # Fetch all cookbooks for the group, rather than the household
54 response = self.group_cookbooks.page_all( 1dehijcfklmognb
55 pagination=q,
56 override=ReadCookBook,
57 )
59 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1dehijcfklmognb
60 return response 1dehijcfklmognb
62 @router.post("", response_model=ReadCookBook, status_code=201) 1a
63 def create_one(self, data: CreateCookBook): 1a
64 data = mapper.cast(data, SaveCookBook, group_id=self.group_id, household_id=self.household_id) 1dehijcfklmgnb
65 cookbook = self.mixins.create_one(data) 1dehijcfklmgnb
67 if cookbook: 67 ↛ 76line 67 didn't jump to line 76 because the condition on line 67 was always true1dehijcfklmgnb
68 self.publish_event( 1dehijcfklmgnb
69 event_type=EventTypes.cookbook_created,
70 document_data=EventCookbookData(operation=EventOperation.create, cookbook_id=cookbook.id),
71 group_id=cookbook.group_id,
72 household_id=cookbook.household_id,
73 message=self.t("notifications.generic-created", name=cookbook.name),
74 )
76 return cookbook 1dehijcfklmgnb
78 @router.put("", response_model=list[ReadCookBook]) 1a
79 def update_many(self, data: list[UpdateCookBook]): 1a
80 updated_by_group_and_household: defaultdict[UUID4, defaultdict[UUID4, list[ReadCookBook]]] = defaultdict(
81 lambda: defaultdict(list)
82 )
84 for cookbook in data: 84 ↛ 85line 84 didn't jump to line 85 because the loop on line 84 never started
85 cb = self.mixins.update_one(cookbook, cookbook.id)
86 updated_by_group_and_household[cb.group_id][cb.household_id].append(cb)
88 all_updated: list[ReadCookBook] = []
89 if updated_by_group_and_household: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 for group_id, household_dict in updated_by_group_and_household.items():
91 for household_id, updated_cookbooks in household_dict.items():
92 all_updated.extend(updated_cookbooks)
93 self.publish_event(
94 event_type=EventTypes.cookbook_updated,
95 document_data=EventCookbookBulkData(
96 operation=EventOperation.update, cookbook_ids=[cb.id for cb in updated_cookbooks]
97 ),
98 group_id=group_id,
99 household_id=household_id,
100 )
102 return all_updated
104 @router.get("/{item_id}", response_model=ReadCookBook) 1a
105 def get_one(self, item_id: UUID4 | str): 1a
106 if isinstance(item_id, UUID): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true1cb
107 match_attr = "id"
108 else:
109 try: 1cb
110 UUID(item_id) 1cb
111 match_attr = "id" 1cb
112 except ValueError:
113 match_attr = "slug"
115 # Allow fetching other households' cookbooks
116 cookbook = self.group_cookbooks.get_one(item_id, match_attr) 1cb
117 if cookbook is None: 1cb
118 raise HTTPException(status_code=404)
120 return cookbook 1cb
122 @router.put("/{item_id}", response_model=ReadCookBook) 1a
123 def update_one(self, item_id: str, data: CreateCookBook): 1a
124 cookbook = self.mixins.update_one(data, item_id) # type: ignore 1pg
125 if cookbook: 125 ↛ 134line 125 didn't jump to line 134 because the condition on line 125 was always true1g
126 self.publish_event( 1g
127 event_type=EventTypes.cookbook_updated,
128 document_data=EventCookbookData(operation=EventOperation.update, cookbook_id=cookbook.id),
129 group_id=cookbook.group_id,
130 household_id=cookbook.household_id,
131 message=self.t("notifications.generic-updated", name=cookbook.name),
132 )
134 return cookbook 1g
136 @router.delete("/{item_id}", response_model=ReadCookBook) 1a
137 def delete_one(self, item_id: str): 1a
138 cookbook = self.mixins.delete_one(item_id) 1decfb
139 if cookbook: 139 ↛ 148line 139 didn't jump to line 148 because the condition on line 139 was always true1decfb
140 self.publish_event( 1decfb
141 event_type=EventTypes.cookbook_deleted,
142 document_data=EventCookbookData(operation=EventOperation.delete, cookbook_id=cookbook.id),
143 group_id=cookbook.group_id,
144 household_id=cookbook.household_id,
145 message=self.t("notifications.generic-deleted", name=cookbook.name),
146 )
148 return cookbook 1decfb