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, AAttachment, QAttachment 

11 

12from rfpy import conf 

13 

14 

15log = logging.getLogger(__name__) 

16 

17 

18class OrphanedAttachmentError(ValueError): 

19 pass 

20 

21 

22def get_sub_path(att) -> str: 

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

24 raise ValueError("Can only determine path for attachments with an ID") 

25 

26 if isinstance(att, ProjectAttachment): 

27 return f"{att.project.id}/PROJECT/{att.id}" 

28 

29 if isinstance(att, IssueAttachment): 

30 return f"{att.issue.project.id}/ISSUE/{att.id}" 

31 

32 if isinstance(att, QAttachment): 

33 return f"QUESTION_ATTACHMENTS/{att.id}" 

34 

35 if isinstance(att, AAttachment): 

36 if att.answer is None: 

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

38 raise OrphanedAttachmentError(m) 

39 return f"{att.answer.issue.project_id}/ANSWER/{att.id}" 

40 

41 raise ValueError(f"Attachment {att} is not a recognised type") 

42 

43 

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

45 atts_dir = root_attachments_dir() 

46 return atts_dir.joinpath(local_path) 

47 

48 

49def root_attachments_dir() -> pathlib.Path: 

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

51 

52 

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

54 """pathlib.Path object for the attachment""" 

55 sub_path = get_sub_path(attachment) 

56 return local_to_full(sub_path) 

57 

58 

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

60 """ 

61 Handles saving a physical attachment file associated with an attachment 

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

63 """ 

64 

65 if not hasattr(readable_object, "read"): 

66 raise ValueError("readable_object must have a read() method") 

67 

68 fpath = get_full_path(attachment) 

69 

70 if fpath.exists() and not overwrite: 

71 raise IOError("{} already exists, will not overwrite".format(fpath)) 

72 

73 if not fpath.parent.exists(): 

74 fpath.parent.mkdir(parents=True) 

75 

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

77 

78 attachment.size_bytes = bytes_written 

79 

80 

81def delete_from_disc(attachment): 

82 fpath = get_full_path(attachment) 

83 fpath.unlink() 

84 log.info("Deleted attachment %s from filepath %s", attachment.filename, fpath) 

85 return fpath 

86 

87 

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

89 try: 

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

91 except IndexError: 

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

93 

94 

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

96 """ 

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

98 been deleted, leaving them orphan. 

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

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

101 

102 Returns a tuple: 

103 (deleted, missing, deleted_row_count) 

104 

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

106 """ 

107 root = root_attachments_dir() 

108 deleted_filenames = set() 

109 missing_filenames = set() 

110 deleted_row_count = 0 

111 

112 for a in session.query(AAttachment.id).filter(AAttachment.answer_id.is_(None)): 

113 try: 

114 att_filename = find_orphan_answer_attachment(root, a.id) 

115 if not dry_run: 

116 att_filename.unlink() 

117 deleted_filenames.add(att_filename) 

118 except FileNotFoundError: 

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

120 if prune_db and not dry_run: 

121 session.commit() 

122 rc = session.query(AAttachment).filter(AAttachment.answer_id.is_(None)).delete() 

123 deleted_row_count = rc 

124 

125 return (deleted_filenames, missing_filenames, deleted_row_count) 

126 

127 

128def save_issue_attachment(session, issue_id, user, cgi_file, description): 

129 ia = IssueAttachment() 

130 ia.filename = cgi_file.filename 

131 ia.guess_set_mimetype(cgi_file.filename) 

132 ia.date_uploaded = datetime.now() 

133 ia.author_id = user.id 

134 ia.org_id = user.org_id 

135 ia.description = description 

136 

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

138 ia.size_bytes = 0 

139 

140 ia.issue_id = issue_id 

141 

142 session.add(ia) 

143 session.flush() 

144 

145 save_to_disc(ia, cgi_file.file) 

146 

147 return ia 

148 

149 

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

151 pa = ProjectAttachment() 

152 pa.filename = cgi_file.filename 

153 pa.guess_set_mimetype(cgi_file.filename) 

154 pa.date_uploaded = datetime.now() 

155 pa.author_id = user.id 

156 pa.org_id = user.org_id 

157 pa.description = description 

158 

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

160 pa.size_bytes = 0 

161 

162 pa.project_id = project_id 

163 

164 session.add(pa) 

165 session.flush() 

166 

167 save_to_disc(pa, cgi_file.file) 

168 

169 return pa 

170 

171 

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

173 fpath = delete_from_disc(attachment) 

174 session.delete(attachment) 

175 session.flush() 

176 return fpath