Coverage for opt/mealie/lib/python3.12/site-packages/mealie/repos/seed/seeders.py: 28%

78 statements  

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

1import json 1a

2import pathlib 1a

3from collections.abc import Generator 1a

4from functools import cached_property 1a

5 

6from mealie.schema.labels import MultiPurposeLabelOut, MultiPurposeLabelSave 1a

7from mealie.schema.recipe.recipe_ingredient import ( 1a

8 IngredientFood, 

9 IngredientUnit, 

10 SaveIngredientFood, 

11 SaveIngredientUnit, 

12) 

13from mealie.services.group_services.labels_service import MultiPurposeLabelService 1a

14 

15from ._abstract_seeder import AbstractSeeder 1a

16from .resources import foods, units 1a

17 

18 

19class MultiPurposeLabelSeeder(AbstractSeeder): 1a

20 @cached_property 1a

21 def service(self): 1a

22 return MultiPurposeLabelService(self.repos) 

23 

24 def get_file(self, locale: str | None = None) -> pathlib.Path: 1a

25 # Get the labels from the foods seed file now 

26 locale_path = self.resources / "foods" / "locales" / f"{locale}.json" 

27 return locale_path if locale_path.exists() else foods.en_US 

28 

29 def get_all_labels(self) -> list[MultiPurposeLabelOut]: 1a

30 return self.repos.group_multi_purpose_labels.get_all() 

31 

32 def load_data(self, locale: str | None = None) -> Generator[MultiPurposeLabelSave, None, None]: 1a

33 file = self.get_file(locale) 

34 

35 current_label_names = {label.name for label in self.get_all_labels()} 

36 # load from the foods locale file and remove any empty strings 

37 seed_label_names = set(filter(None, json.loads(file.read_text(encoding="utf-8")).keys())) # type: set[str] 

38 # only seed new labels 

39 to_seed_labels = seed_label_names - current_label_names 

40 for label in to_seed_labels: 

41 yield MultiPurposeLabelSave( 

42 name=label, 

43 group_id=self.repos.group_id, 

44 ) 

45 

46 def seed(self, locale: str | None = None) -> None: 1a

47 self.logger.info("Seeding MultiPurposeLabel") 

48 for label in self.load_data(locale): 

49 try: 

50 self.service.create_one(label) 

51 except Exception as e: 

52 self.logger.error(e) 

53 

54 

55class IngredientUnitsSeeder(AbstractSeeder): 1a

56 def get_file(self, locale: str | None = None) -> pathlib.Path: 1a

57 locale_path = self.resources / "units" / "locales" / f"{locale}.json" 

58 return locale_path if locale_path.exists() else units.en_US 

59 

60 def get_all_units(self) -> list[IngredientUnit]: 1a

61 return self.repos.ingredient_units.get_all() 

62 

63 def load_data(self, locale: str | None = None) -> Generator[SaveIngredientUnit, None, None]: 1a

64 file = self.get_file(locale) 

65 

66 seen_unit_names = {unit.name for unit in self.get_all_units()} 

67 for unit in json.loads(file.read_text(encoding="utf-8")).values(): 

68 if unit["name"] in seen_unit_names: 

69 continue 

70 

71 seen_unit_names.add(unit["name"]) 

72 yield SaveIngredientUnit( 

73 group_id=self.repos.group_id, 

74 name=unit["name"], 

75 plural_name=unit.get("plural_name"), 

76 description=unit["description"], 

77 abbreviation=unit["abbreviation"], 

78 plural_abbreviation=unit.get("plural_abbreviation"), 

79 ) 

80 

81 def seed(self, locale: str | None = None) -> None: 1a

82 self.logger.info("Seeding Ingredient Units") 

83 for unit in self.load_data(locale): 

84 try: 

85 self.repos.ingredient_units.create(unit) 

86 except Exception as e: 

87 self.logger.error(e) 

88 

89 

90class IngredientFoodsSeeder(AbstractSeeder): 1a

91 def get_file(self, locale: str | None = None) -> pathlib.Path: 1a

92 locale_path = self.resources / "foods" / "locales" / f"{locale}.json" 

93 return locale_path if locale_path.exists() else foods.en_US 

94 

95 def get_label(self, value: str) -> MultiPurposeLabelOut | None: 1a

96 return self.repos.group_multi_purpose_labels.get_one(value, "name") 

97 

98 def get_all_foods(self) -> list[IngredientFood]: 1a

99 return self.repos.ingredient_foods.get_all() 

100 

101 def load_data(self, locale: str | None = None) -> Generator[SaveIngredientFood, None, None]: 1a

102 file = self.get_file(locale) 

103 

104 # get all current unique foods 

105 seen_foods_names = {food.name for food in self.get_all_foods()} 

106 for label, values in json.loads(file.read_text(encoding="utf-8")).items(): 

107 label_out = self.get_label(label) 

108 

109 for food_name, attributes in values["foods"].items(): 

110 if food_name in seen_foods_names: 

111 continue 

112 

113 seen_foods_names.add(food_name) 

114 yield SaveIngredientFood( 

115 group_id=self.repos.group_id, 

116 name=attributes["name"], 

117 plural_name=attributes.get("plural_name"), 

118 description="", # description expected to be empty string by UnitFoodBase class 

119 label_id=label_out.id if label_out and label_out.id else None, 

120 ) 

121 

122 def seed(self, locale: str | None = None) -> None: 1a

123 self.logger.info("Seeding Ingredient Foods") 

124 for food in self.load_data(locale): 

125 try: 

126 self.repos.ingredient_foods.create(food) 

127 except Exception as e: 

128 self.logger.error(e)