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

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 8 ctx1decifgha

17 self.user = user 8 ctx1decifgha

18 self.group = group 8 ctx1decifgha

19 super().__init__() 8 ctx1decifgha

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) 2 ctx1ia

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

24 

25 exporter.run(self.repos) 2 ctx1ia

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 ctx1fa

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

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

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

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: 3 ctx1gha

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

74 

75 if recipe is None: 3 ctx1gha

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

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

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

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

88 

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

90 for slug in recipes: 3 ctx1dea

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

92 

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

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

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 ctx1ca

109 try: 2 ctx1ca

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

111 except Exception as e: 2 ctx1ca

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

113 self.logger.error(e) 2 ctx1ca