Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_mealplan.py: 96%
74 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1from datetime import date 1a
2from functools import cached_property 1a
4from dateutil.tz import tzlocal 1a
5from fastapi import APIRouter, Depends, HTTPException 1a
7from mealie.core.exceptions import mealie_registered_exceptions 1a
8from mealie.repos.all_repositories import get_repositories 1a
9from mealie.repos.repository_meals import RepositoryMeals 1a
10from mealie.routes._base import controller 1a
11from mealie.routes._base.base_controllers import BaseCrudController 1a
12from mealie.routes._base.mixins import HttpRepo 1a
13from mealie.schema import mapper 1a
14from mealie.schema.meal_plan import CreatePlanEntry, ReadPlanEntry, SavePlanEntry, UpdatePlanEntry 1a
15from mealie.schema.meal_plan.new_meal import CreateRandomEntry, PlanEntryPagination, PlanEntryType 1a
16from mealie.schema.meal_plan.plan_rules import PlanRulesDay 1a
17from mealie.schema.recipe.recipe import Recipe 1a
18from mealie.schema.response.pagination import PaginationQuery 1a
19from mealie.schema.response.responses import ErrorResponse 1a
20from mealie.services.event_bus_service.event_types import EventMealplanCreatedData, EventTypes 1a
22router = APIRouter(prefix="/households/mealplans", tags=["Households: Mealplans"]) 1a
25@controller(router) 1a
26class GroupMealplanController(BaseCrudController): 1a
27 @cached_property 1a
28 def repo(self) -> RepositoryMeals: 1a
29 return self.repos.meals 1pdiclfmehjnokgb
31 def registered_exceptions(self, ex: type[Exception]) -> str: 1a
32 registered = { 1cb
33 **mealie_registered_exceptions(self.translator),
34 }
35 return registered.get(ex, self.t("generic.server-error")) 1cb
37 @cached_property 1a
38 def mixins(self): 1a
39 return HttpRepo[CreatePlanEntry, ReadPlanEntry, UpdatePlanEntry]( 1pdicfejkgb
40 self.repo,
41 self.logger,
42 self.registered_exceptions,
43 )
45 def _get_random_recipes_from_mealplan( 1a
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.
52 Recipes from all households are included unless the rules specify a household filter.
53 """
55 rules = self.repos.group_meal_plan_rules.get_rules(PlanRulesDay.from_date(plan_date), entry_type.value) 1diclfmehjnokgb
56 cross_household_recipes = get_repositories( 1diclfmehjnokgb
57 self.session, group_id=self.group_id, household_id=None
58 ).recipes.by_user(self.user.id)
60 qf_string = " AND ".join([f"({rule.query_filter_string})" for rule in rules if rule.query_filter_string]) 1diclfmehjnokgb
61 recipes_data = cross_household_recipes.page_all( 1diclfmehjnokgb
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
72 @router.get("", response_model=PlanEntryPagination) 1a
73 def get_all( 1a
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: 1diclfmehjnkgb
81 if not start_date: 1dclfehgb
82 date_filter = f"date <= {end_date}" 1dclfhgb
84 elif not end_date: 1dcehb
85 date_filter = f"date >= {start_date}" 1eb
87 else:
88 date_filter = f"date >= {start_date} AND date <= {end_date}" 1dcehb
90 if q.query_filter: 1dclfehgb
91 q.query_filter = f"({q.query_filter}) AND ({date_filter})" 1cfehb
93 else:
94 q.query_filter = date_filter 1dclhgb
96 return self.repo.page_all(pagination=q) 1diclfmehjnkgb
98 @router.post("", response_model=ReadPlanEntry, status_code=201) 1a
99 def create_one(self, data: CreatePlanEntry): 1a
100 data = mapper.cast(data, SavePlanEntry, group_id=self.group_id, user_id=self.user.id) 1dicfejkgb
101 result = self.mixins.create_one(data) 1dicfejkgb
103 self.publish_event( 1dicfejkgb
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 )
117 return result 1dicfejkgb
119 @router.get("/today") 1a
120 def get_todays_meals(self): 1a
121 local_tz = tzlocal()
122 return self.repo.get_today(tz=local_tz)
124 @router.post("/random", response_model=ReadPlanEntry) 1a
125 def create_random_meal(self, data: CreateRandomEntry): 1a
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.
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) 1diclfmehjnokgb
135 if not random_recipes: 135 ↛ 140line 135 didn't jump to line 140 because the condition on line 135 was always true
136 raise HTTPException(
137 status_code=404, detail=ErrorResponse.respond(message=self.t("mealplan.no-recipes-match-your-rules"))
138 )
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 )
151 @router.get("/{item_id}", response_model=ReadPlanEntry) 1a
152 def get_one(self, item_id: int): 1a
153 return self.mixins.get_one(item_id) 1fb
155 @router.put("/{item_id}", response_model=ReadPlanEntry) 1a
156 def update_one(self, item_id: int, data: UpdatePlanEntry): 1a
157 return self.mixins.update_one(data, item_id) 1pb
159 @router.delete("/{item_id}", response_model=ReadPlanEntry) 1a
160 def delete_one(self, item_id: int): 1a
161 return self.mixins.delete_one(item_id) 1cb