Coverage for postrfp/buyer/api/endpoints/attachments.py: 99%

1""" 

2Manage file attachments and uploads 

3""" 

4 

5from sqlalchemy.orm import Session 

6 

7from postrfp.model.questionnaire.qelements import ( 

8 QuestionAttachment, 

9) 

10from postrfp.shared import attachments, fetch, serial 

11from postrfp.shared.decorators import http 

12from postrfp.authorisation import perms 

13from postrfp.buyer.api import authorise 

14from postrfp.shared.response import XAccelAttachmentResponse, XAccelResponse 

15from postrfp.model import QuestionInstance, User 

16from postrfp.model.audit import AuditEvent, evt_types 

17 

18 

19safo = serial.Attachment.model_validate 

20 

21 

22@http 

23def get_project_attachments( 

24 session: Session, user: User, project_id: int 

25) -> list[serial.Attachment]: 

26 """ 

27 list details for all Project Attachments (attachments uploaded by the Project author - buyer) 

28 """ 

29 project = fetch.project(session, project_id) 

30 authorise.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

31 return [safo(att) for att in project.list_attachments(user)] 

32 

33 

34@http 

35def get_project_issueattachments( 

36 session: Session, user: User, project_id: int 

37) -> list[serial.IssueAttachment]: 

38 """ 

39 list all Issue attachments (uploaded by Respondents) for the current Project 

40 """ 

41 project = fetch.project(session, project_id) 

42 authorise.check( 

43 user, 

44 perms.ISSUE_VIEW_ANSWERS, 

45 project=project, 

46 section_id=project.section_id, 

47 deny_restricted=False, 

48 ) 

49 return [ 

50 serial.IssueAttachment.model_validate(att) 

51 for att in project.list_issue_attachments(user) 

52 ] 

53 

54 

55@http 

56def get_project_answerattachments( 

57 session: Session, user: User, project_id: int 

58) -> list[serial.AnswerAttachment]: 

59 """ 

60 list Answer Attachments for the current project - attachments uploaded as answers to 

61 questions with File Upload question elements 

62 """ 

63 project = fetch.project(session, project_id) 

64 authorise.check( 

65 user, 

66 perms.ISSUE_VIEW_ANSWERS, 

67 project=project, 

68 deny_restricted=False, 

69 section_id=project.section_id, 

70 ) 

71 return [ 

72 serial.AnswerAttachment.model_validate(att) 

73 for att in fetch.answer_attachments_q(project, user) 

74 ] 

75 

76 

77@http 

78def get_project_attachment( 

79 session: Session, user: User, project_id: int, attachment_id: int 

80) -> XAccelResponse: 

81 """ 

82 Download the Project Attachment with the given ID 

83 """ 

84 project = fetch.project(session, project_id) 

85 authorise.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

86 attachment = project.get_attachment(user, attachment_id) 

87 

88 return XAccelAttachmentResponse(attachment) 

89 

90 

91@http 

92def post_project_attachment( 

93 session: Session, 

94 user: User, 

95 project_id: int, 

96 attachment_upload: str, 

97 attachment_description: str, 

98) -> serial.Id: 

99 """ 

100 Upload a file as an attachment to the project. 

101 

102 This action can be performed for Projects at status Draft or Live 

103 """ 

104 

105 project = fetch.project(session, project_id) 

106 authorise.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

107 attachment = attachments.save_project_attachment( 

108 session, project_id, user, attachment_upload, attachment_description 

109 ) 

110 session.flush() 

111 changes = [("filename", attachment.filename, "")] 

112 evt = AuditEvent.create( 

113 session, 

114 evt_types.PROJECT_ATTACHMENT_ADDED, 

115 project=project, 

116 user=user, 

117 object_id=attachment.id, 

118 change_list=changes, 

119 ) 

120 session.add(evt) 

121 return serial.Id(id=attachment.id) 

122 

123 

124@http 

125def delete_project_attachment( 

126 session: Session, user: User, project_id: int, attachment_id: int 

127): 

128 """ 

129 Delete the Project Attachment with the given ID 

130 

131 This action can be performed for Projects at status Draft or Live 

132 """ 

133 project = fetch.project(session, project_id) 

134 authorise.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

135 attachment = project.get_attachment(user, attachment_id) 

136 filename = attachment.filename 

137 attachments.delete_project_attachment(session, attachment) 

138 changes = [("filename", "", filename)] 

139 evt = AuditEvent.create( 

140 session, 

141 evt_types.PROJECT_ATTACHMENT_REMOVED, 

142 project=project, 

143 user=user, 

144 object_id=attachment_id, 

145 change_list=changes, 

146 ) 

147 session.add(evt) 

148 

149 

150@http 

151def get_project_issue_attachment( 

152 session: Session, user: User, project_id: int, issue_id: int, attachment_id: int 

153) -> XAccelResponse: 

154 """ 

155 Download the Issue Attachment with the given ID 

156 """ 

157 project = fetch.project(session, project_id) 

158 issue = fetch.issue(session, issue_id) 

159 authorise.check( 

160 user, 

161 perms.ISSUE_VIEW_ANSWERS, 

162 issue=issue, 

163 project=project, 

164 section_id=project.section_id, 

165 ) 

166 attachment = issue.get_attachment(attachment_id) 

167 

168 return XAccelAttachmentResponse(attachment) 

169 

170 

171@http 

172def get_question_element_attachment( 

173 session: Session, user: User, question_id: int, element_id: int 

174) -> XAccelResponse: 

175 """ 

176 Download the Question Attachment associated with the given Element ID 

177 """ 

178 q_instance = session.get_one(QuestionInstance, question_id) 

179 q_element = session.get_one(QuestionAttachment, element_id) 

180 

181 if q_element.question_id != q_instance.question_def_id: 

182 raise ValueError(f"Element {element_id} is not part of Question {question_id}") 

183 authorise.check( 

184 user, 

185 perms.PROJECT_VIEW_QUESTIONNAIRE, 

186 project=q_instance.project, 

187 section_id=q_instance.section_id, 

188 ) 

189 return XAccelAttachmentResponse(q_element.attachment) 

190 

191 

192@http 

193def get_answer_attachment( 

194 session: Session, user: User, answer_id: int 

195) -> XAccelResponse: 

196 """ 

197 Download the Answer Attachment with the given ID 

198 """ 

199 answer = fetch.answer(session, answer_id) 

200 issue = answer.issue 

201 authorise.check( 

202 user, 

203 perms.ISSUE_VIEW_ANSWERS, 

204 issue=issue, 

205 project=issue.project, 

206 section_id=answer.question_instance.section_id, 

207 ) 

208 

209 if not answer.attachment: 

210 raise ValueError(f"No attachment found for answer {answer_id}") 

211 

212 return XAccelAttachmentResponse(answer.attachment)