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 17:29 +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 1EJmgBnhijuofpqdrwkscleCtDxAyb
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 1zvmKgnhijuofpqdrGwkHLscleItxyb
37 def registered_exceptions(self, ex: type[Exception]) -> str: 1a
38 registered = { 1EgBhijufdklCDb
39 **mealie_registered_exceptions(self.translator),
40 }
41 return registered.get(ex, self.t("generic.server-error")) 1EgBhijufdklCDb
43 @cached_property 1a
44 def mixins(self): 1a
45 return HttpRepo[CreateCookBook, ReadCookBook, UpdateCookBook]( 1EJmgBnhijuofpqdrwkscleCtDxAyb
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( 1mKgnhijuofpqdrGwkHLscleItxyb
55 pagination=q,
56 override=ReadCookBook,
57 )
59 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1mgnhijuofpqdrGkHscleItb
60 return response 1mgnhijuofpqdrGkHscleItb
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) 1mgBnhijuofpqdrwksclCtDxAyb
65 cookbook = self.mixins.create_one(data) 1mgBnhijuofpqdrwksclCtDxAyb
67 if cookbook: 67 ↛ 76line 67 didn't jump to line 76 because the condition on line 67 was always true1mgnhijofpqdrwkscltxAyb
68 self.publish_event( 1mgnhijofpqdrwkscltxAyb
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 1mgnhijofpqdrwkscltxAyb
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( 1Fb
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 started1Fb
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] = [] 1Fb
89 if updated_by_group_and_household: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true1Fb
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 1Fb
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 true1zvceb
107 match_attr = "id"
108 else:
109 try: 1zvceb
110 UUID(item_id) 1zvceb
111 match_attr = "id" 1ceb
112 except ValueError: 1zvb
113 match_attr = "slug" 1zvb
115 # Allow fetching other households' cookbooks
116 cookbook = self.group_cookbooks.get_one(item_id, match_attr) 1zvceb
117 if cookbook is None: 1zvceb
118 raise HTTPException(status_code=404) 1zveb
120 return cookbook 1vcb
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 1Jdb
125 if cookbook: 125 ↛ 134line 125 didn't jump to line 134 because the condition on line 125 was always true1d
126 self.publish_event( 1d
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 1d
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) 1Efeb
139 if cookbook: 139 ↛ 148line 139 didn't jump to line 148 because the condition on line 139 was always true1eb
140 self.publish_event( 1eb
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 1eb