Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_group_recipe_actions.py: 69%
55 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
1from functools import cached_property 1a
3import requests 1a
4from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException, status 1a
5from fastapi.encoders import jsonable_encoder 1a
6from pydantic import UUID4 1a
8from mealie.core.exceptions import NoEntryFound 1a
9from mealie.routes._base.base_controllers import BaseUserController 1a
10from mealie.routes._base.controller import controller 1a
11from mealie.routes._base.mixins import HttpRepo 1a
12from mealie.schema.household.group_recipe_action import ( 1a
13 CreateGroupRecipeAction,
14 GroupRecipeActionOut,
15 GroupRecipeActionPagination,
16 GroupRecipeActionPayload,
17 GroupRecipeActionType,
18 SaveGroupRecipeAction,
19)
20from mealie.schema.response import ErrorResponse 1a
21from mealie.schema.response.pagination import PaginationQuery 1a
22from mealie.services.recipe.recipe_service import RecipeService 1a
24router = APIRouter(prefix="/households/recipe-actions", tags=["Households: Recipe Actions"]) 1a
27@controller(router) 1a
28class GroupRecipeActionController(BaseUserController): 1a
29 @cached_property 1a
30 def repo(self): 1a
31 return self.repos.group_recipe_actions 1lcfdeghijkb
33 @property 1a
34 def mixins(self): 1a
35 return HttpRepo[CreateGroupRecipeAction, GroupRecipeActionOut, SaveGroupRecipeAction](self.repo, self.logger) 1cfdeghijkb
37 # ==================================================================================================================
38 # CRUD
40 @router.get("", response_model=GroupRecipeActionPagination) 1a
41 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 1a
42 response = self.repo.page_all( 1lcfdeghijkb
43 pagination=q,
44 override=GroupRecipeActionOut,
45 )
47 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 1lcfdeghijb
48 return response 1lcfdeghijb
50 @router.post("", response_model=GroupRecipeActionOut, status_code=201) 1a
51 def create_one(self, data: CreateGroupRecipeAction): 1a
52 save = data.cast(SaveGroupRecipeAction, group_id=self.group_id, household_id=self.household_id) 1cfdeghijkb
53 return self.mixins.create_one(save) 1cfdeghijkb
55 @router.get("/{item_id}", response_model=GroupRecipeActionOut) 1a
56 def get_one(self, item_id: UUID4): 1a
57 return self.mixins.get_one(item_id) 1db
59 @router.put("/{item_id}", response_model=GroupRecipeActionOut) 1a
60 def update_one(self, item_id: UUID4, data: SaveGroupRecipeAction): 1a
61 return self.mixins.update_one(data, item_id)
63 @router.delete("/{item_id}", response_model=GroupRecipeActionOut) 1a
64 def delete_one(self, item_id: UUID4): 1a
65 return self.mixins.delete_one(item_id) 1ceb
67 # ==================================================================================================================
68 # Actions
70 @router.post("/{item_id}/trigger/{recipe_slug}", status_code=202) 1a
71 def trigger_action( 1a
72 self, item_id: UUID4, recipe_slug: str, bg_tasks: BackgroundTasks, recipe_scale: float = Body(1, embed=True)
73 ) -> None:
74 recipe_action = self.repos.group_recipe_actions.get_one(item_id)
75 if not recipe_action:
76 raise HTTPException(
77 status.HTTP_404_NOT_FOUND,
78 detail=ErrorResponse.respond(message="Not found."),
79 )
81 if recipe_action.action_type == GroupRecipeActionType.post.value:
82 task_action = requests.post
83 else:
84 raise HTTPException(
85 status.HTTP_400_BAD_REQUEST,
86 detail=ErrorResponse.respond(message=f'Cannot trigger action type "{recipe_action.action_type}".'),
87 )
89 recipe_service = RecipeService(self.repos, self.user, self.household, translator=self.translator)
90 try:
91 recipe = recipe_service.get_one(recipe_slug)
92 except NoEntryFound as e:
93 raise HTTPException(
94 status.HTTP_404_NOT_FOUND,
95 detail=ErrorResponse.respond(message="Not found."),
96 ) from e
98 payload = GroupRecipeActionPayload(action=recipe_action, content=recipe, recipe_scale=recipe_scale)
99 bg_tasks.add_task(
100 task_action,
101 url=recipe_action.url,
102 json=jsonable_encoder(payload.model_dump()),
103 timeout=15,
104 )