Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/recipe/bulk_actions.py: 96%

45 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 03:04 +0000

1from functools import cached_property 

2from pathlib import Path 

3 

4from fastapi import APIRouter, HTTPException 

5 

6from mealie.core.dependencies.dependencies import get_temporary_zip_path 

7from mealie.core.security import create_file_token 

8from mealie.routes._base import BaseUserController, controller 

9from mealie.schema.group.group_exports import GroupDataExport 

10from mealie.schema.recipe.recipe_bulk_actions import ( 

11 AssignCategories, 

12 AssignSettings, 

13 AssignTags, 

14 DeleteRecipes, 

15 ExportRecipes, 

16) 

17from mealie.schema.response.responses import SuccessResponse 

18from mealie.services.recipe.recipe_bulk_service import RecipeBulkActionsService 

19 

20router = APIRouter(prefix="/bulk-actions") 

21 

22 

23@controller(router) 

24class RecipeBulkActionsController(BaseUserController): 

25 @cached_property 

26 def service(self) -> RecipeBulkActionsService: 

27 return RecipeBulkActionsService(self.repos, self.user, self.group) 

28 

29 # TODO Should these actions return some success response? 

30 @router.post("/tag") 

31 def bulk_tag_recipes(self, tag_data: AssignTags): 

32 self.service.assign_tags(tag_data.recipes, tag_data.tags) 

33 

34 @router.post("/settings") 

35 def bulk_settings_recipes(self, settings_data: AssignSettings): 

36 self.service.set_settings(settings_data.recipes, settings_data.settings) 

37 

38 @router.post("/categorize") 

39 def bulk_categorize_recipes(self, assign_cats: AssignCategories): 

40 self.service.assign_categories(assign_cats.recipes, assign_cats.categories) 

41 

42 @router.post("/delete") 

43 def bulk_delete_recipes(self, delete_recipes: DeleteRecipes): 

44 self.service.delete_recipes(delete_recipes.recipes) 

45 

46 @router.post("/export", status_code=202) 

47 def bulk_export_recipes(self, export_recipes: ExportRecipes): 

48 with get_temporary_zip_path() as temp_path: 

49 self.service.export_recipes(temp_path, export_recipes.recipes) 

50 

51 @router.get("/export/download") 

52 def get_exported_data_token(self, path: Path): 

53 """Returns a token to download a file""" 

54 path = Path(path).resolve() 

55 

56 if not path.is_relative_to(self.folders.DATA_DIR): 56 ↛ 59line 56 didn't jump to line 59 because the condition on line 56 was always true

57 raise HTTPException(400, "path must be relative to data directory") 

58 

59 return {"fileToken": create_file_token(path)} 

60 

61 @router.get("/export", response_model=list[GroupDataExport]) 

62 def get_exported_data(self): 

63 return self.service.get_exports() 

64 

65 @router.delete("/export/purge", response_model=SuccessResponse) 

66 def purge_export_data(self): 

67 """Remove all exports data, including items on disk without database entry""" 

68 amountDelete = self.service.purge_exports() 

69 return SuccessResponse.respond(f"{amountDelete} exports deleted")