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

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 4 ctx1edca

17 self.user = user 4 ctx1edca

18 self.group = group 4 ctx1edca

19 super().__init__() 4 ctx1edca

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

29 

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

31 all_exports = self.get_exports() (empty)

32 

33 exports_deleted = 0 (empty)

34 for export in all_exports: 34 ↛ 35line 34 didn't jump to line 35 because the loop on line 34 never started(empty)

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

44 

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

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

49 if match.is_file(): 

50 match.unlink() 

51 exports_deleted += 1 

52 

53 return exports_deleted (empty)

54 

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

56 for slug in recipes: 2 ctx1ca

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

58 

59 if recipe is None or recipe.settings is None: 2 ctx1ca

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

61 

62 settings.locked = recipe.settings.locked (empty)

63 recipe.settings = settings (empty)

64 

65 try: (empty)

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

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: 2 ctx1ca

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

74 

75 if recipe is None: 2 ctx1ca

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

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

79 recipe.tags = [] 

80 

81 recipe.tags += tags # type: ignore (empty)

82 

83 try: (empty)

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

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: 2 ctx1ea

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

92 

93 if recipe is None: 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was always true2 ctx1ea

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

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: 2 ctx1da

109 try: 2 ctx1da

110 self.repos.recipes.delete(slug) 2 ctx1da

111 except Exception as e: 2 ctx1da

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

113 self.logger.error(e) 2 ctx1da