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

1from pathlib import Path 1 ctx1b

2 

3from mealie.core.exceptions import UnexpectedNone 1 ctx1b

4from mealie.repos.repository_factory import AllRepositories 1 ctx1b

5from mealie.schema.group.group_exports import GroupDataExport 1 ctx1b

6from mealie.schema.recipe import CategoryBase 1 ctx1b

7from mealie.schema.recipe.recipe_category import TagBase 1 ctx1b

8from mealie.schema.recipe.recipe_settings import RecipeSettings 1 ctx1b

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

10from mealie.services._base_service import BaseService 1 ctx1b

11from mealie.services.exporter import Exporter, RecipeExporter 1 ctx1b

12 

13 

14class RecipeBulkActionsService(BaseService): 1 ctx1b

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

16 self.repos = repos 2 ctx1ca

17 self.user = user 2 ctx1ca

18 self.group = group 2 ctx1ca

19 super().__init__() 2 ctx1ca

20 

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

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

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

24 

25 exporter.run(self.repos) (empty)

26 

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

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

29 

30 def purge_exports(self) -> int: 1 ctx1b

31 all_exports = self.get_exports() 1 ctx1c

32 

33 exports_deleted = 0 1 ctx1c

34 for export in all_exports: 34 ↛ 35line 34 didn't jump to line 35 because the loop on line 34 never started1 ctx1c

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) 1 ctx1c

44 

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

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

49 if match.is_file(): 

50 match.unlink() 

51 exports_deleted += 1 

52 

53 return exports_deleted 1 ctx1c

54 

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

56 for slug in recipes: (empty)

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

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(empty)

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

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

72 for slug in recipes: (empty)

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

74 

75 if recipe is None: 75 ↛ 78line 75 didn't jump to line 78 because the condition on line 75 was always true(empty)

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

77 

78 if recipe.tags is None: 

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

90 for slug in recipes: (empty)

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

92 

93 if recipe is None: 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was always true(empty)

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

95 

96 if recipe.recipe_category is None: 

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

108 for slug in recipes: (empty)

109 try: (empty)

110 self.repos.recipes.delete(slug) (empty)

111 except Exception as e: (empty)

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

113 self.logger.error(e) (empty)