Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/recipe/timeline_events.py: 60%
75 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 17:29 +0000
1import shutil 1a
2from functools import cached_property 1a
4from fastapi import Depends, File, Form, HTTPException 1a
5from pydantic import UUID4 1a
7from mealie.repos.all_repositories import get_repositories 1a
8from mealie.routes._base import BaseCrudController, controller 1a
9from mealie.routes._base.mixins import HttpRepo 1a
10from mealie.routes._base.routers import MealieCrudRoute, UserAPIRouter 1a
11from mealie.schema.recipe.recipe_timeline_events import ( 1a
12 RecipeTimelineEventCreate,
13 RecipeTimelineEventIn,
14 RecipeTimelineEventOut,
15 RecipeTimelineEventPagination,
16 RecipeTimelineEventUpdate,
17 TimelineEventImage,
18)
19from mealie.schema.recipe.request_helpers import UpdateImageResponse 1a
20from mealie.schema.response.pagination import PaginationQuery 1a
21from mealie.services import urls 1a
22from mealie.services.event_bus_service.event_types import EventOperation, EventRecipeTimelineEventData, EventTypes 1a
23from mealie.services.recipe.recipe_data_service import RecipeDataService 1a
25router = UserAPIRouter(route_class=MealieCrudRoute, prefix="/timeline/events") 1a
28@controller(router) 1a
29class RecipeTimelineEventsController(BaseCrudController): 1a
30 @cached_property 1a
31 def repo(self): 1a
32 return self.repos.recipe_timeline_events 1defghuvwijklmncopqrstb
34 @cached_property 1a
35 def group_recipes(self): 1a
36 return get_repositories(self.session, group_id=self.group_id, household_id=None).recipes
38 @cached_property 1a
39 def mixins(self): 1a
40 return HttpRepo[RecipeTimelineEventCreate, RecipeTimelineEventOut, RecipeTimelineEventUpdate]( 1cb
41 self.repo,
42 self.logger,
43 self.registered_exceptions,
44 )
46 @router.get("", response_model=RecipeTimelineEventPagination) 1a
47 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a
48 response = self.repo.page_all( 1defghuvwijklmncopqrstb
49 pagination=q,
50 override=RecipeTimelineEventOut,
51 )
53 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1defghijklmncopqrstb
54 return response 1defghijklmncopqrstb
56 @router.post("", response_model=RecipeTimelineEventOut, status_code=201) 1a
57 def create_one(self, data: RecipeTimelineEventIn): 1a
58 # if the user id is not specified, use the currently-authenticated user
59 data.user_id = data.user_id or self.user.id
61 recipe = self.group_recipes.get_one(data.recipe_id, "id")
62 if not recipe:
63 raise HTTPException(status_code=404, detail="recipe not found")
65 event_data = data.cast(RecipeTimelineEventCreate)
66 event = self.mixins.create_one(event_data)
68 self.publish_event(
69 event_type=EventTypes.recipe_updated,
70 document_data=EventRecipeTimelineEventData(
71 operation=EventOperation.create, recipe_slug=recipe.slug, recipe_timeline_event_id=event.id
72 ),
73 group_id=recipe.group_id,
74 household_id=recipe.household_id,
75 message=self.t(
76 "notifications.generic-updated-with-url",
77 name=recipe.name,
78 url=urls.recipe_url(self.group.slug, recipe.slug, self.settings.BASE_URL),
79 ),
80 )
82 return event
84 @router.get("/{item_id}", response_model=RecipeTimelineEventOut) 1a
85 def get_one(self, item_id: UUID4): 1a
86 return self.mixins.get_one(item_id) 1c
88 @router.put("/{item_id}", response_model=RecipeTimelineEventOut) 1a
89 def update_one(self, item_id: UUID4, data: RecipeTimelineEventUpdate): 1a
90 event = self.mixins.patch_one(data, item_id)
91 recipe = self.group_recipes.get_one(event.recipe_id, "id")
92 if recipe:
93 self.publish_event(
94 event_type=EventTypes.recipe_updated,
95 document_data=EventRecipeTimelineEventData(
96 operation=EventOperation.update, recipe_slug=recipe.slug, recipe_timeline_event_id=event.id
97 ),
98 group_id=recipe.group_id,
99 household_id=recipe.household_id,
100 message=self.t(
101 "notifications.generic-updated-with-url",
102 name=recipe.name,
103 url=urls.recipe_url(self.group.slug, recipe.slug, self.settings.BASE_URL),
104 ),
105 )
107 return event
109 @router.delete("/{item_id}", response_model=RecipeTimelineEventOut) 1a
110 def delete_one(self, item_id: UUID4): 1a
111 event = self.mixins.delete_one(item_id)
112 if event.image_dir.exists(): 112 ↛ 118line 112 didn't jump to line 118 because the condition on line 112 was always true
113 try:
114 shutil.rmtree(event.image_dir)
115 except FileNotFoundError:
116 pass
118 recipe = self.group_recipes.get_one(event.recipe_id, "id")
119 if recipe: 119 ↛ 134line 119 didn't jump to line 134 because the condition on line 119 was always true
120 self.publish_event(
121 event_type=EventTypes.recipe_updated,
122 document_data=EventRecipeTimelineEventData(
123 operation=EventOperation.delete, recipe_slug=recipe.slug, recipe_timeline_event_id=event.id
124 ),
125 group_id=recipe.group_id,
126 household_id=recipe.household_id,
127 message=self.t(
128 "notifications.generic-updated-with-url",
129 name=recipe.name,
130 url=urls.recipe_url(self.group.slug, recipe.slug, self.settings.BASE_URL),
131 ),
132 )
134 return event
136 # ==================================================================================================================
137 # Image and Assets
139 @router.put("/{item_id}/image", response_model=UpdateImageResponse) 1a
140 def update_event_image(self, item_id: UUID4, image: bytes = File(...), extension: str = Form(...)): 1a
141 event = self.mixins.get_one(item_id)
142 data_service = RecipeDataService(event.recipe_id)
143 data_service.write_image(image, extension, event.image_dir)
145 if event.image != TimelineEventImage.has_image.value:
146 event.image = TimelineEventImage.has_image
147 event = self.mixins.patch_one(event.cast(RecipeTimelineEventUpdate), event.id)
148 recipe = self.group_recipes.get_one(event.recipe_id, "id")
149 if recipe:
150 self.publish_event(
151 event_type=EventTypes.recipe_updated,
152 document_data=EventRecipeTimelineEventData(
153 operation=EventOperation.update, recipe_slug=recipe.slug, recipe_timeline_event_id=event.id
154 ),
155 group_id=recipe.group_id,
156 household_id=recipe.household_id,
157 message=self.t(
158 "notifications.generic-updated-with-url",
159 name=recipe.name,
160 url=urls.recipe_url(self.group.slug, recipe.slug, self.settings.BASE_URL),
161 ),
162 )
164 return UpdateImageResponse(image=TimelineEventImage.has_image.value)