Coverage for app/logic/fileHandler.py: 80%

1import os 

2from flask import redirect, url_for 

3from app import app 

4from app.models.attachmentUpload import AttachmentUpload 

5from app.models.cceMinorProposal import CCEMinorProposal 

6from app.models.program import Program 

7import glob 

8 

9class FileHandler: 

10 def __init__(self, files=None, courseId=None, eventId=None, programId=None, proposalId=None): 

11 self.files = files 

12 if not isinstance(self.files, list): 

13 self.files = [self.files] 

14 self.path = app.config['files']['base_path'] 

15 self.courseId = courseId 

16 self.eventId = eventId 

17 self.programId = programId 

18 self.proposalId = proposalId 

19 

20 if courseId: 

21 self.path = os.path.join(self.path, app.config['files']['course_attachment_path'], str(courseId)) 

22 elif eventId: 

23 self.path = os.path.join(self.path, app.config['files']['event_attachment_path']) 

24 elif programId: 

25 self.path = os.path.join(self.path, app.config['files']['program_attachment_path']) 

26 elif proposalId: 

27 self.path = os.path.join(self.path, app.config['files']['proposal_attachment_path'], str(proposalId)) 

28 

29 def makeDirectory(self): 

30 try: 

31 extraDir = "" 

32 if self.eventId: 

33 extraDir = str(self.eventId) 

34 

35 os.makedirs(os.path.join(self.path, extraDir)) 

36 except OSError as e: 

37 if e.errno != 17: 

38 print(f'Fail to create directory: {e}') 

39 raise e 

40 

41 def getFileFullPath(self, newfilename=''): 

42 try: 

43 if self.eventId or self.courseId or self.programId or self.proposalId: 

44 filePath = (os.path.join(self.path, newfilename)) 

45 except AttributeError: 

46 pass 

47 except FileExistsError: 

48 pass 

49 

50 return filePath 

51 

52 def saveFiles(self, parentEvent=None): 

53 """ 

54 Saves attachments for different types and creates DB record for stored attachment 

55 """ 

56 for file in self.files: 

57 saveFileToFilesystem = None 

58 

59 if self.eventId: 

60 attachmentName = str(parentEvent.id) + "/" + file.filename 

61 fileObject, created = AttachmentUpload.get_or_create( 

62 event_id = self.eventId, 

63 fileName = attachmentName 

64 ) 

65 

66 if created and parentEvent and parentEvent.id == self.eventId: 

67 saveFileToFilesystem = attachmentName 

68 

69 elif self.courseId or self.proposalId: 

70 recordId = self.courseId if self.courseId else self.proposalId 

71 fieldName = 'course' if self.courseId else 'proposal' 

72 fileData = { 

73 fieldName: recordId, 

74 "fileName": file.filename 

75 } 

76 fileObject, created = AttachmentUpload.get_or_create(**fileData) 

77 saveFileToFilesystem = file.filename if created else None 

78 

79 elif self.programId: 

80 

81 # remove the existing file 

82 deleteFileObject = AttachmentUpload.get_or_none(program=self.programId) 

83 if deleteFileObject: 

84 self.deleteFile(deleteFileObject.id) 

85 

86 # add the new file 

87 fileType = file.filename.split('.')[-1] 

88 fileName = f"{self.programId}.{fileType}" 

89 AttachmentUpload.create(program=self.programId, fileName=fileName) 

90 currentProgramID = fileName 

91 saveFileToFilesystem = currentProgramID 

92 

93 else: 

94 saveFileToFilesystem = file.filename 

95 

96 if saveFileToFilesystem: 

97 self.makeDirectory() 

98 file.save(self.getFileFullPath(newfilename=saveFileToFilesystem)) 

99 

100 def retrievePath(self, files): 

101 pathDict = {} 

102 for file in files: 

103 pathDict[file.fileName] = ((self.path + "/" + file.fileName)[3:], file) 

104 return pathDict 

105 

106 def deleteFile(self, fileId): 

107 file = AttachmentUpload.get_by_id(fileId) 

108 file.delete_instance() 

109 if not AttachmentUpload.select().where(AttachmentUpload.fileName == file.fileName).exists(): 

110 path = os.path.join(self.path, file.fileName) 

111 os.remove(path) 

112 

113 def changeDisplay(self, fileId, isDisplayed): 

114 file = AttachmentUpload.get_by_id(fileId) 

115 

116 # Uncheck all other checkboxes for the same event 

117 AttachmentUpload.update(isDisplayed=False).where(AttachmentUpload.event == file.event).execute() 

118 

119 # Check the selected checkbox 

120 AttachmentUpload.update(isDisplayed=isDisplayed).where(AttachmentUpload.id == fileId).execute() 

121 return ""