Coverage for rfpy/api/attachments.py: 100%

1import pathlib 

2from typing import BinaryIO 

3 

4import logging 

5from datetime import datetime 

6from rfpy.model.meta import AttachmentMixin 

7 

8from sqlalchemy.orm.session import Session 

9 

10from rfpy.model import (ProjectAttachment, IssueAttachment, 

11 AAttachment, QAttachment) 

12 

13from rfpy import conf 

14 

15 

16log = logging.getLogger(__name__) 

17 

18 

19class OrphanedAttachmentError(ValueError): 

20 pass 

21 

22 

23def get_sub_path(att) -> str: 

24 

25 if not isinstance(att.id, int): 

26 raise ValueError('Can only determine path for attachments with an ID') 

27 

28 if isinstance(att, ProjectAttachment): 

29 return f'{att.project.id}/PROJECT/{att.id}' 

30 

31 if isinstance(att, IssueAttachment): 

32 return f'{att.issue.project.id}/ISSUE/{att.id}' 

33 

34 if isinstance(att, QAttachment): 

35 return f'QUESTION_ATTACHMENTS/{att.id}' 

36 

37 if isinstance(att, AAttachment): 

38 if att.answer is None: 

39 m = f'answer attachment # {att.id} is an orphan- no Answer found' 

40 raise OrphanedAttachmentError(m) 

41 return f'{att.answer.issue.project_id}/ANSWER/{att.id}' 

42 

43 raise ValueError(f'Attachment {att} is not a recognised type') 

44 

45 

46def local_to_full(local_path) -> pathlib.Path: 

47 atts_dir = root_attachments_dir() 

48 return atts_dir.joinpath(local_path) 

49 

50 

51def root_attachments_dir() -> pathlib.Path: 

52 return pathlib.Path(conf.CONF.attachments_dir) 

53 

54 

55def get_full_path(attachment) -> pathlib.Path: 

56 '''pathlib.Path object for the attachment''' 

57 sub_path = get_sub_path(attachment) 

58 return local_to_full(sub_path) 

59 

60 

61def save_to_disc(attachment, readable_object: BinaryIO, overwrite=False): 

62 ''' 

63 Handles saving a physical attachment file associated with an attachment 

64 object/ record (e.g. a Project Attachment) to disc. 

65 ''' 

66 

67 if not hasattr(readable_object, 'read'): 

68 raise ValueError('readable_object must have a read() method') 

69 

70 fpath = get_full_path(attachment) 

71 

72 if fpath.exists() and not overwrite: 

73 raise IOError('{} already exists, will not overwrite'.format(fpath)) 

74 

75 if not fpath.parent.exists(): 

76 fpath.parent.mkdir(parents=True) 

77 

78 bytes_written = fpath.write_bytes(readable_object.read()) 

79 

80 attachment.size_bytes = bytes_written 

81 

82 

83def delete_from_disc(attachment): 

84 fpath = get_full_path(attachment) 

85 fpath.unlink() 

86 log.info("Deleted attachment %s from filepath %s", 

87 attachment.filename, fpath) 

88 return fpath 

89 

90 

91def find_orphan_answer_attachment(root_dir: pathlib.Path, attachment_id): 

92 try: 

93 return list(root_dir.glob(f'*/ANSWER/{attachment_id}'))[0] 

94 except IndexError: 

95 raise FileNotFoundError(f"Couldn't find attachment id {attachment_id}") 

96 

97 

98def delete_orphan_answer_attachments(session, dry_run=True, prune_db=False): 

99 ''' 

100 Find and delete answer attachment files whose parent answer record has 

101 been deleted, leaving them orphan. 

102 If dry_run is True, just identify and return the files that would be chopped 

103 If prune_db is True, also delete the records from the database. 

104 

105 Returns a tuple: 

106 (deleted, missing, deleted_row_count) 

107 

108 First two values are a Set of pathlib.Path objects 

109 ''' 

110 root = root_attachments_dir() 

111 deleted_filenames = set() 

112 missing_filenames = set() 

113 deleted_row_count = 0 

114 

115 for a in session.query(AAttachment.id)\ 

116 .filter(AAttachment.answer_id.is_(None)): 

117 try: 

118 att_filename = find_orphan_answer_attachment(root, a.id) 

119 if not dry_run: 

120 att_filename.unlink() 

121 deleted_filenames.add(att_filename) 

122 except FileNotFoundError: 

123 missing_filenames.add(f"/ANSWER/{a.id}") 

124 if prune_db and not dry_run: 

125 session.commit() 

126 rc = session.query(AAttachment)\ 

127 .filter(AAttachment.answer_id.is_(None))\ 

128 .delete() 

129 deleted_row_count = rc 

130 

131 return (deleted_filenames, missing_filenames, deleted_row_count) 

132 

133 

134def save_issue_attachment(session, issue_id, user, cgi_file, 

135 description): 

136 

137 ia = IssueAttachment() 

138 ia.filename = cgi_file.filename 

139 ia.guess_set_mimetype(cgi_file.filename) 

140 ia.date_uploaded = datetime.now() 

141 ia.author_id = user.id 

142 ia.org_id = user.org_id 

143 ia.description = description 

144 

145 # Necessary to set size to non-null value for Mysql to save the record 

146 ia.size_bytes = 0 

147 

148 ia.issue_id = issue_id 

149 

150 session.add(ia) 

151 session.flush() 

152 

153 save_to_disc(ia, cgi_file.file) 

154 

155 return ia 

156 

157 

158def save_project_attachment(session, project_id, user, cgi_file, description): 

159 

160 pa = ProjectAttachment() 

161 pa.filename = cgi_file.filename 

162 pa.guess_set_mimetype(cgi_file.filename) 

163 pa.date_uploaded = datetime.now() 

164 pa.author_id = user.id 

165 pa.org_id = user.org_id 

166 pa.description = description 

167 

168 # Necessary to set size to non-null value for Mysql to save the record 

169 pa.size_bytes = 0 

170 

171 pa.project_id = project_id 

172 

173 session.add(pa) 

174 session.flush() 

175 

176 save_to_disc(pa, cgi_file.file) 

177 

178 return pa 

179 

180 

181def delete_project_attachment(session: Session, attachment: AttachmentMixin): 

182 fpath = delete_from_disc(attachment) 

183 session.delete(attachment) 

184 session.flush() 

185 return fpath