Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/admin/admin_backups.py: 68%

1import operator 

2import shutil 

3from pathlib import Path 

4 

5from fastapi import APIRouter, File, HTTPException, UploadFile, status 

6 

7from mealie.core.config import get_app_dirs 

8from mealie.core.root_logger import get_logger 

9from mealie.core.security import create_file_token 

10from mealie.pkgs.stats.fs_stats import pretty_size 

11from mealie.routes._base import BaseAdminController, controller 

12from mealie.schema.admin.backup import AllBackups, BackupFile 

13from mealie.schema.response.responses import ErrorResponse, FileTokenResponse, SuccessResponse 

14from mealie.services.backups_v2.backup_v2 import BackupSchemaMismatch, BackupV2 

15 

16logger = get_logger() 

17router = APIRouter(prefix="/backups") 

18 

19 

20@controller(router) 

21class AdminBackupController(BaseAdminController): 

22 def _backup_path(self, name) -> Path: 

23 return get_app_dirs().BACKUP_DIR / name 

24 

25 @router.get("", response_model=AllBackups) 

26 def get_all(self): 

27 app_dirs = get_app_dirs() 

28 imports = [] 

29 for archive in app_dirs.BACKUP_DIR.glob("*.zip"): 

30 backup = BackupFile( 

31 name=archive.name, date=archive.stat().st_mtime, size=pretty_size(archive.stat().st_size) 

32 ) 

33 imports.append(backup) 

34 

35 templates = [template.name for template in app_dirs.TEMPLATE_DIR.glob("*.*")] 

36 imports.sort(key=operator.attrgetter("date"), reverse=True) 

37 

38 return AllBackups(imports=imports, templates=templates) 

39 

40 @router.post("", status_code=status.HTTP_201_CREATED, response_model=SuccessResponse) 

41 def create_one(self): 

42 backup = BackupV2() 

43 

44 try: 

45 backup.backup() 

46 except Exception as e: 

47 logger.exception(e) 

48 raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR) from e 

49 

50 return SuccessResponse.respond("Backup created successfully") 

51 

52 @router.get("/{file_name}", response_model=FileTokenResponse) 

53 def get_one(self, file_name: str): 

54 """Returns a token to download a file""" 

55 file = self._backup_path(file_name) 

56 

57 if not file.exists(): 

58 raise HTTPException(status.HTTP_404_NOT_FOUND) 

59 

60 return FileTokenResponse.respond(create_file_token(file)) 

61 

62 @router.delete("/{file_name}", status_code=status.HTTP_200_OK, response_model=SuccessResponse) 

63 def delete_one(self, file_name: str): 

64 file = self._backup_path(file_name) 

65 

66 if not file.is_file(): 

67 raise HTTPException(status.HTTP_400_BAD_REQUEST) 

68 try: 

69 file.unlink() 

70 except Exception as e: 

71 raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR) from e 

72 

73 return SuccessResponse.respond(f"{file_name} has been deleted.") 

74 

75 @router.post("/upload", response_model=SuccessResponse) 

76 def upload_one(self, archive: UploadFile = File(...)): 

77 """Upload a .zip File to later be imported into Mealie""" 

78 if "." not in archive.filename: 

79 raise HTTPException(status.HTTP_400_BAD_REQUEST) 

80 

81 if archive.filename.split(".")[-1] != "zip": 81 ↛ 84line 81 didn't jump to line 84 because the condition on line 81 was always true

82 raise HTTPException(status.HTTP_400_BAD_REQUEST) 

83 

84 name = Path(archive.filename).stem 

85 

86 app_dirs = get_app_dirs() 

87 dest = app_dirs.BACKUP_DIR.joinpath(f"{name}.zip") 

88 

89 if dest.absolute().parent != app_dirs.BACKUP_DIR: 

90 raise HTTPException(status.HTTP_400_BAD_REQUEST) 

91 

92 with dest.open("wb") as buffer: 

93 shutil.copyfileobj(archive.file, buffer) 

94 

95 if not dest.is_file(): 

96 raise HTTPException(status.HTTP_400_BAD_REQUEST) 

97 return SuccessResponse.respond("Upload successful") 

98 

99 @router.post("/{file_name}/restore", response_model=SuccessResponse) 

100 def import_one(self, file_name: str): 

101 backup = BackupV2() 

102 

103 file = self._backup_path(file_name) 

104 

105 try: 

106 backup.restore(file) 

107 except BackupSchemaMismatch as e: 

108 raise HTTPException( 

109 status.HTTP_400_BAD_REQUEST, 

110 ErrorResponse.respond("database backup schema version does not match current database"), 

111 ) from e 

112 except Exception as e: 

113 logger.exception(e) 

114 raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR) from e 

115 

116 return SuccessResponse.respond("Restore successful")