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

1from collections.abc import Iterable 1 ctx1a

2from typing import cast 1 ctx1a

3from uuid import UUID 1 ctx1a

4 

5from pydantic import UUID4 1 ctx1a

6from slugify import slugify 1 ctx1a

7from sqlalchemy import func, select 1 ctx1a

8from sqlalchemy.exc import IntegrityError 1 ctx1a

9 

10from mealie.db.models._model_base import SqlAlchemyBase 1 ctx1a

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

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

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

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

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

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

17from mealie.repos.repository_generic import GroupRepositoryGeneric, HouseholdRepositoryGeneric 1 ctx1a

18from mealie.schema.household import ( 1 ctx1a

19 HouseholdCreate, 

20 HouseholdInDB, 

21 HouseholdRecipeOut, 

22 HouseholdStatistics, 

23 UpdateHousehold, 

24) 

25 

26 

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

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

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

30 data = data.model_dump() 2 ctx1ab

31 

32 if not data.get("group_id"): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true2 ctx1ab

33 data["group_id"] = self.group_id 

34 max_attempts = 10 2 ctx1ab

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

36 

37 attempts = 0 2 ctx1ab

38 while True: 2 ctx1ab

39 try: 2 ctx1ab

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

41 return super().create(data) 2 ctx1ab

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]: 1 ctx1a

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: 1 ctx1a

55 if isinstance(new_data, HouseholdCreate): 

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]: 1 ctx1a

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: 1 ctx1a

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

71 raise Exception("group_id not set") 

72 dbhousehold = ( 2 ctx1ab

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 true2 ctx1ab

78 return None 

79 return self.schema.model_validate(dbhousehold) 2 ctx1ab

80 

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

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

83 try: 1 ctx1c

84 slug_or_id = UUID(slug_or_id) 1 ctx1c

85 except ValueError: 1 ctx1c

86 pass 1 ctx1c

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 true1 ctx1c

89 return self.get_one(slug_or_id) 

90 else: 

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

92 

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

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

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

96 if filter_household: 1 ctx1c

97 stmt = stmt.filter_by(household_id=household_id) 1 ctx1c

98 return self.session.scalar(stmt) 1 ctx1c

99 

100 return HouseholdStatistics( 1 ctx1c

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]): 1 ctx1a

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

113 if not self.household_id: 

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)