Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/recipe/recipe_bulk_service.py: 79%

1from pathlib import Path 

2 

3from mealie.core.exceptions import UnexpectedNone 

4from mealie.repos.repository_factory import AllRepositories 

5from mealie.schema.group.group_exports import GroupDataExport 

6from mealie.schema.recipe import CategoryBase 

7from mealie.schema.recipe.recipe_category import TagBase 

8from mealie.schema.recipe.recipe_settings import RecipeSettings 

9from mealie.schema.user.user import GroupInDB, PrivateUser 

10from mealie.services._base_service import BaseService 

11from mealie.services.exporter import Exporter, RecipeExporter 

12 

13 

14class RecipeBulkActionsService(BaseService): 

15 def __init__(self, repos: AllRepositories, user: PrivateUser, group: GroupInDB): 

16 self.repos = repos 

17 self.user = user 

18 self.group = group 

19 super().__init__() 

20 

21 def export_recipes(self, temp_path: Path, slugs: list[str]) -> None: 

22 recipe_exporter = RecipeExporter(self.repos, self.group.id, slugs) 

23 exporter = Exporter(self.group.id, temp_path, [recipe_exporter]) 

24 

25 exporter.run(self.repos) 

26 

27 def get_exports(self) -> list[GroupDataExport]: 

28 return self.repos.group_exports.multi_query({"group_id": self.group.id}) 

29 

30 def purge_exports(self) -> int: 

31 all_exports = self.get_exports() 

32 

33 exports_deleted = 0 

34 for export in all_exports: 

35 try: 

36 Path(export.path).unlink(missing_ok=True) 

37 self.repos.group_exports.delete(export.id) 

38 exports_deleted += 1 

39 except Exception as e: 

40 self.logger.error(f"Failed to delete export {export.id}") 

41 self.logger.error(e) 

42 

43 group = self.repos.groups.get_one(self.group.id) 

44 

45 if group is None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 raise UnexpectedNone("Failed to purge exports for group, no group found") 

47 

48 for match in group.directory.glob("**/export/*zip"): 48 ↛ 49line 48 didn't jump to line 49 because the loop on line 48 never started

49 if match.is_file(): 

50 match.unlink() 

51 exports_deleted += 1 

52 

53 return exports_deleted 

54 

55 def set_settings(self, recipes: list[str], settings: RecipeSettings) -> None: 

56 for slug in recipes: 

57 recipe = self.repos.recipes.get_one(slug) 

58 

59 if recipe is None or recipe.settings is None: 59 ↛ 62line 59 didn't jump to line 62 because the condition on line 59 was always true

60 raise UnexpectedNone(f"Failed to set settings for recipe {slug}, no recipe found") 

61 

62 settings.locked = recipe.settings.locked 

63 recipe.settings = settings 

64 

65 try: 

66 self.repos.recipes.update(slug, recipe) 

67 except Exception as e: 

68 self.logger.error(f"Failed to set settings for recipe {slug}") 

69 self.logger.error(e) 

70 

71 def assign_tags(self, recipes: list[str], tags: list[TagBase]) -> None: 

72 for slug in recipes: 

73 recipe = self.repos.recipes.get_one(slug) 

74 

75 if recipe is None: 

76 raise UnexpectedNone(f"Failed to tag recipe {slug}, no recipe found") 

77 

78 if recipe.tags is None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 recipe.tags = [] 

80 

81 recipe.tags += tags # type: ignore 

82 

83 try: 

84 self.repos.recipes.update(slug, recipe) 

85 except Exception as e: 

86 self.logger.error(f"Failed to tag recipe {slug}") 

87 self.logger.error(e) 

88 

89 def assign_categories(self, recipes: list[str], categories: list[CategoryBase]) -> None: 

90 for slug in recipes: 

91 recipe = self.repos.recipes.get_one(slug) 

92 

93 if recipe is None: 

94 raise UnexpectedNone(f"Failed to categorize recipe {slug}, no recipe found") 

95 

96 if recipe.recipe_category is None: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true

97 recipe.recipe_category = [] 

98 

99 recipe.recipe_category += categories # type: ignore 

100 

101 try: 

102 self.repos.recipes.update(slug, recipe) 

103 except Exception as e: 

104 self.logger.error(f"Failed to categorize recipe {slug}") 

105 self.logger.error(e) 

106 

107 def delete_recipes(self, recipes: list[str]) -> None: 

108 for slug in recipes: 

109 try: 

110 self.repos.recipes.delete(slug) 

111 except Exception as e: 

112 self.logger.error(f"Failed to delete recipe {slug}") 

113 self.logger.error(e)