Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_mealplan.py: 100%

1from datetime import date 

2from functools import cached_property 

3 

4from dateutil.tz import tzlocal 

5from fastapi import APIRouter, Depends, HTTPException 

6 

7from mealie.core.exceptions import mealie_registered_exceptions 

8from mealie.repos.all_repositories import get_repositories 

9from mealie.repos.repository_meals import RepositoryMeals 

10from mealie.routes._base import controller 

11from mealie.routes._base.base_controllers import BaseCrudController 

12from mealie.routes._base.mixins import HttpRepo 

13from mealie.schema import mapper 

14from mealie.schema.meal_plan import CreatePlanEntry, ReadPlanEntry, SavePlanEntry, UpdatePlanEntry 

15from mealie.schema.meal_plan.new_meal import CreateRandomEntry, PlanEntryPagination, PlanEntryType 

16from mealie.schema.meal_plan.plan_rules import PlanRulesDay 

17from mealie.schema.recipe.recipe import Recipe 

18from mealie.schema.response.pagination import PaginationQuery 

19from mealie.schema.response.responses import ErrorResponse 

20from mealie.services.event_bus_service.event_types import EventMealplanCreatedData, EventTypes 

21 

22router = APIRouter(prefix="/households/mealplans", tags=["Households: Mealplans"]) 

23 

24 

25@controller(router) 

26class GroupMealplanController(BaseCrudController): 

27 @cached_property 

28 def repo(self) -> RepositoryMeals: 

29 return self.repos.meals 

30 

31 def registered_exceptions(self, ex: type[Exception]) -> str: 

32 registered = { 

33 **mealie_registered_exceptions(self.translator), 

34 } 

35 return registered.get(ex, self.t("generic.server-error")) 

36 

37 @cached_property 

38 def mixins(self): 

39 return HttpRepo[CreatePlanEntry, ReadPlanEntry, UpdatePlanEntry]( 

40 self.repo, 

41 self.logger, 

42 self.registered_exceptions, 

43 ) 

44 

45 def _get_random_recipes_from_mealplan( 

46 self, plan_date: date, entry_type: PlanEntryType, limit: int = 1 

47 ) -> list[Recipe]: 

48 """ 

49 Gets rules for a mealplan and returns a list of random recipes based on the rules. 

50 May return zero recipes if no recipes match the filter criteria. 

51 

52 Recipes from all households are included unless the rules specify a household filter. 

53 """ 

54 

55 rules = self.repos.group_meal_plan_rules.get_rules(PlanRulesDay.from_date(plan_date), entry_type.value) 

56 cross_household_recipes = get_repositories( 

57 self.session, group_id=self.group_id, household_id=None 

58 ).recipes.by_user(self.user.id) 

59 

60 qf_string = " AND ".join([f"({rule.query_filter_string})" for rule in rules if rule.query_filter_string]) 

61 recipes_data = cross_household_recipes.page_all( 

62 pagination=PaginationQuery( 

63 page=1, 

64 per_page=limit, 

65 query_filter=qf_string, 

66 order_by="random", 

67 pagination_seed=self.repo._random_seed(), 

68 ) 

69 ) 

70 return recipes_data.items 

71 

72 @router.get("", response_model=PlanEntryPagination) 

73 def get_all( 

74 self, 

75 q: PaginationQuery = Depends(PaginationQuery), 

76 start_date: date | None = None, 

77 end_date: date | None = None, 

78 ): 

79 # merge start and end dates into pagination query only if either is provided 

80 if start_date or end_date: 

81 if not start_date: 

82 date_filter = f"date <= {end_date}" 

83 

84 elif not end_date: 

85 date_filter = f"date >= {start_date}" 

86 

87 else: 

88 date_filter = f"date >= {start_date} AND date <= {end_date}" 

89 

90 if q.query_filter: 

91 q.query_filter = f"({q.query_filter}) AND ({date_filter})" 

92 

93 else: 

94 q.query_filter = date_filter 

95 

96 return self.repo.page_all(pagination=q) 

97 

98 @router.post("", response_model=ReadPlanEntry, status_code=201) 

99 def create_one(self, data: CreatePlanEntry): 

100 data = mapper.cast(data, SavePlanEntry, group_id=self.group_id, user_id=self.user.id) 

101 result = self.mixins.create_one(data) 

102 

103 self.publish_event( 

104 event_type=EventTypes.mealplan_entry_created, 

105 document_data=EventMealplanCreatedData( 

106 mealplan_id=result.id, 

107 recipe_id=data.recipe_id, 

108 recipe_name=result.recipe.name if result.recipe else None, 

109 recipe_slug=result.recipe.slug if result.recipe else None, 

110 date=data.date, 

111 ), 

112 group_id=result.group_id, 

113 household_id=result.household_id, 

114 message=f"Mealplan entry created for {data.date} for {data.entry_type}", 

115 ) 

116 

117 return result 

118 

119 @router.get("/today") 

120 def get_todays_meals(self): 

121 local_tz = tzlocal() 

122 return self.repo.get_today(tz=local_tz) 

123 

124 @router.post("/random", response_model=ReadPlanEntry) 

125 def create_random_meal(self, data: CreateRandomEntry): 

126 """ 

127 `create_random_meal` is a route that provides the randomized functionality for mealplaners. 

128 It operates by following the rules set out in the household's mealplan settings. If no settings 

129 are set, it will return any random meal. 

130 

131 Refer to the mealplan settings routes for more information on how rules can be applied 

132 to the random meal selector. 

133 """ 

134 random_recipes = self._get_random_recipes_from_mealplan(data.date, data.entry_type) 

135 if not random_recipes: 

136 raise HTTPException( 

137 status_code=404, detail=ErrorResponse.respond(message=self.t("mealplan.no-recipes-match-your-rules")) 

138 ) 

139 

140 recipe = random_recipes[0] 

141 return self.mixins.create_one( 

142 SavePlanEntry( 

143 date=data.date, 

144 entry_type=data.entry_type, 

145 recipe_id=recipe.id, 

146 group_id=self.group_id, 

147 user_id=self.user.id, 

148 ) 

149 ) 

150 

151 @router.get("/{item_id}", response_model=ReadPlanEntry) 

152 def get_one(self, item_id: int): 

153 return self.mixins.get_one(item_id) 

154 

155 @router.put("/{item_id}", response_model=ReadPlanEntry) 

156 def update_one(self, item_id: int, data: UpdatePlanEntry): 

157 return self.mixins.update_one(data, item_id) 

158 

159 @router.delete("/{item_id}", response_model=ReadPlanEntry) 

160 def delete_one(self, item_id: int): 

161 return self.mixins.delete_one(item_id)