Coverage for opt/mealie/lib/python3.12/site-packages/mealie/repos/repository_household.py: 85%

1from collections.abc import Iterable 

2from typing import cast 

3from uuid import UUID 

4 

5from pydantic import UUID4 

6from slugify import slugify 

7from sqlalchemy import func, select 

8from sqlalchemy.exc import IntegrityError 

9 

10from mealie.db.models._model_base import SqlAlchemyBase 

11from mealie.db.models.household import Household, HouseholdToRecipe 

12from mealie.db.models.recipe.category import Category 

13from mealie.db.models.recipe.recipe import RecipeModel 

14from mealie.db.models.recipe.tag import Tag 

15from mealie.db.models.recipe.tool import Tool 

16from mealie.db.models.users.users import User 

17from mealie.repos.repository_generic import GroupRepositoryGeneric, HouseholdRepositoryGeneric 

18from mealie.schema.household import ( 

19 HouseholdCreate, 

20 HouseholdInDB, 

21 HouseholdRecipeOut, 

22 HouseholdStatistics, 

23 UpdateHousehold, 

24) 

25 

26 

27class RepositoryHousehold(GroupRepositoryGeneric[HouseholdInDB, Household]): 

28 def create(self, data: HouseholdCreate | dict) -> HouseholdInDB: 

29 if isinstance(data, HouseholdCreate): 29 ↛ 32line 29 didn't jump to line 32 because the condition on line 29 was always true

30 data = data.model_dump() 

31 

32 if not data.get("group_id"): 

33 data["group_id"] = self.group_id 

34 max_attempts = 10 

35 original_name = cast(str, data["name"]) 

36 

37 attempts = 0 

38 while True: 

39 try: 

40 data["slug"] = slugify(data["name"]) 

41 return super().create(data) 

42 except IntegrityError: 

43 self.session.rollback() 

44 attempts += 1 

45 if attempts >= max_attempts: 

46 raise 

47 

48 data["name"] = f"{original_name} ({attempts})" 

49 

50 def create_many(self, data: Iterable[HouseholdInDB | dict]) -> list[HouseholdInDB]: 

51 # since create uses special logic for resolving slugs, we don't want to use the standard create_many method 

52 return [self.create(new_household) for new_household in data] 

53 

54 def update(self, match_value: str | int | UUID4, new_data: UpdateHousehold | dict) -> HouseholdInDB: 

55 if isinstance(new_data, HouseholdCreate): 55 ↛ 58line 55 didn't jump to line 58 because the condition on line 55 was always true

56 new_data.slug = slugify(new_data.name) 

57 else: 

58 new_data["slug"] = slugify(new_data["name"]) 

59 

60 return super().update(match_value, new_data) 

61 

62 def update_many(self, data: Iterable[UpdateHousehold | dict]) -> list[HouseholdInDB]: 

63 # since update uses special logic for resolving slugs, we don't want to use the standard update_many method 

64 return [ 

65 self.update(household["id"] if isinstance(household, dict) else household.id, household) 

66 for household in data 

67 ] 

68 

69 def get_by_name(self, name: str) -> HouseholdInDB | None: 

70 if not self.group_id: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 raise Exception("group_id not set") 

72 dbhousehold = ( 

73 self.session.execute(select(self.model).filter_by(name=name, group_id=self.group_id)) 

74 .scalars() 

75 .one_or_none() 

76 ) 

77 if dbhousehold is None: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 return None 

79 return self.schema.model_validate(dbhousehold) 

80 

81 def get_by_slug_or_id(self, slug_or_id: str | UUID) -> HouseholdInDB | None: 

82 if isinstance(slug_or_id, str): 82 ↛ 88line 82 didn't jump to line 88 because the condition on line 82 was always true

83 try: 

84 slug_or_id = UUID(slug_or_id) 

85 except ValueError: 

86 pass 

87 

88 if isinstance(slug_or_id, UUID): 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 return self.get_one(slug_or_id) 

90 else: 

91 return self.get_one(slug_or_id, key="slug") 

92 

93 def statistics(self, group_id: UUID4, household_id: UUID4) -> HouseholdStatistics: 

94 def model_count(model: type[SqlAlchemyBase], *, filter_household: bool = True) -> int: 

95 stmt = select(func.count(model.id)).filter_by(group_id=group_id) 

96 if filter_household: 

97 stmt = stmt.filter_by(household_id=household_id) 

98 return self.session.scalar(stmt) 

99 

100 return HouseholdStatistics( 

101 # household-level statistics 

102 total_recipes=model_count(RecipeModel), 

103 total_users=model_count(User), 

104 # group-level statistics 

105 total_categories=model_count(Category, filter_household=False), 

106 total_tags=model_count(Tag, filter_household=False), 

107 total_tools=model_count(Tool, filter_household=False), 

108 ) 

109 

110 

111class RepositoryHouseholdRecipes(HouseholdRepositoryGeneric[HouseholdRecipeOut, HouseholdToRecipe]): 

112 def get_by_recipe(self, recipe_id: UUID4) -> HouseholdRecipeOut | None: 

113 if not self.household_id: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 raise Exception("household_id not set") 

115 

116 stmt = select(HouseholdToRecipe).filter( 

117 HouseholdToRecipe.household_id == self.household_id, HouseholdToRecipe.recipe_id == recipe_id 

118 ) 

119 result = self.session.execute(stmt).scalars().one_or_none() 

120 return None if result is None else self.schema.model_validate(result)