Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/household_services/household_service.py: 67%

61 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 17:29 +0000

1from uuid import UUID 1a

2 

3from pydantic import UUID4 1a

4 

5from mealie.core import exceptions 1a

6from mealie.repos.all_repositories import get_repositories 1a

7from mealie.repos.repository_factory import AllRepositories 1a

8from mealie.schema.household import HouseholdCreate, HouseholdRecipeSummary 1a

9from mealie.schema.household.household import HouseholdRecipeCreate, HouseholdRecipeUpdate 1a

10from mealie.schema.household.household_preferences import CreateHouseholdPreferences, SaveHouseholdPreferences 1a

11from mealie.schema.household.household_statistics import HouseholdStatistics 1a

12from mealie.schema.recipe.recipe import Recipe 1a

13from mealie.services._base_service import BaseService 1a

14 

15 

16class HouseholdService(BaseService): 1a

17 def __init__(self, group_id: UUID4, household_id: UUID4, repos: AllRepositories): 1a

18 self.group_id = group_id 1cb

19 self.household_id = household_id 1cb

20 self.repos = repos 1cb

21 super().__init__() 1cb

22 

23 def _get_recipe(self, recipe_slug: str | UUID) -> Recipe | None: 1a

24 key = "id" 

25 if not isinstance(recipe_slug, UUID): 25 ↛ 31line 25 didn't jump to line 31 because the condition on line 25 was always true

26 try: 

27 UUID(recipe_slug) 

28 except ValueError: 

29 key = "slug" 

30 

31 cross_household_recipes = get_repositories( 

32 self.repos.session, group_id=self.group_id, household_id=None 

33 ).recipes 

34 return cross_household_recipes.get_one(recipe_slug, key) 

35 

36 @staticmethod 1a

37 def create_household( 1a

38 repos: AllRepositories, h_base: HouseholdCreate, prefs: CreateHouseholdPreferences | None = None 

39 ): 

40 new_household = repos.households.create(h_base) 1ad

41 if prefs is None: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true1ad

42 group = repos.groups.get_one(new_household.group_id) 

43 if group and group.preferences: 

44 prefs = CreateHouseholdPreferences( 

45 private_household=group.preferences.private_group, 

46 recipe_public=not group.preferences.private_group, 

47 ) 

48 else: 

49 prefs = CreateHouseholdPreferences() 

50 save_prefs = prefs.cast(SaveHouseholdPreferences, household_id=new_household.id) 1ad

51 

52 household_repos = get_repositories( 1ad

53 repos.session, group_id=new_household.group_id, household_id=new_household.id 

54 ) 

55 household_repos.household_preferences.create(save_prefs) 1ad

56 return new_household 1ad

57 

58 def calculate_statistics( 1a

59 self, group_id: UUID4 | None = None, household_id: UUID4 | None = None 

60 ) -> HouseholdStatistics: 

61 """ 

62 calculate_statistics calculates the statistics for the group and returns 

63 a HouseholdStatistics object. 

64 """ 

65 group_id = group_id or self.group_id 1c

66 household_id = household_id or self.household_id 1c

67 

68 return self.repos.households.statistics(group_id, household_id) 1c

69 

70 def get_household_recipe(self, recipe_slug: str) -> HouseholdRecipeSummary | None: 1a

71 """Returns recipe data for the current household""" 

72 recipe = self._get_recipe(recipe_slug) 

73 if not recipe: 73 ↛ 76line 73 didn't jump to line 76 because the condition on line 73 was always true

74 return None 

75 

76 household_recipe_out = self.repos.household_recipes.get_by_recipe(recipe.id) 

77 if household_recipe_out: 

78 return household_recipe_out.cast(HouseholdRecipeSummary) 

79 else: 

80 return HouseholdRecipeSummary(recipe_id=recipe.id) 

81 

82 def set_household_recipe(self, recipe_slug: str | UUID, data: HouseholdRecipeUpdate) -> HouseholdRecipeSummary: 1a

83 """Sets the household's recipe data""" 

84 recipe = self._get_recipe(recipe_slug) 

85 if not recipe: 85 ↛ 88line 85 didn't jump to line 88 because the condition on line 85 was always true

86 raise exceptions.NoEntryFound("Recipe not found.") 

87 

88 existing_household_recipe = self.repos.household_recipes.get_by_recipe(recipe.id) 

89 

90 if existing_household_recipe: 

91 updated_data = existing_household_recipe.cast(HouseholdRecipeUpdate, **data.model_dump()) 

92 household_recipe_out = self.repos.household_recipes.patch(existing_household_recipe.id, updated_data) 

93 else: 

94 create_data = HouseholdRecipeCreate( 

95 household_id=self.household_id, recipe_id=recipe.id, **data.model_dump() 

96 ) 

97 household_recipe_out = self.repos.household_recipes.create(create_data) 

98 

99 return household_recipe_out.cast(HouseholdRecipeSummary)