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

1""" 

2Manage file attachments and uploads 

3""" 

4 

5from typing import List 

6 

7from rfpy.suxint import http 

8from rfpy.auth import perms 

9from rfpy.api import fetch, validate, attachments 

10from rfpy.web.response import XAccelAttachmentResponse, XAccelResponse 

11from rfpy.web import serial 

12from rfpy.model import QuestionInstance 

13from rfpy.model.audit import AuditEvent, evt_types 

14 

15 

16safo = serial.Attachment.model_validate 

17 

18 

19@http 

20def get_project_attachments(session, user, project_id) -> List[serial.Attachment]: 

21 """ 

22 List details for all Project Attachments (attachments uploaded by the Project author - buyer) 

23 """ 

24 project = fetch.project(session, project_id) 

25 validate.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

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

27 

28 

29@http 

30def get_project_issueattachments( 

31 session, user, project_id 

32) -> List[serial.IssueAttachment]: 

33 """ 

34 List all Issue attachments (uploaded by Respondents) for the current Project 

35 """ 

36 project = fetch.project(session, project_id) 

37 validate.check( 

38 user, 

39 perms.ISSUE_VIEW_ANSWERS, 

40 project=project, 

41 section_id=project.section_id, 

42 deny_restricted=False, 

43 ) 

44 return [ 

45 serial.IssueAttachment.model_validate(att) 

46 for att in project.list_issue_attachments(user) 

47 ] 

48 

49 

50@http 

51def get_project_answerattachments( 

52 session, user, project_id 

53) -> List[serial.AnswerAttachment]: 

54 """ 

55 List Answer Attachments for the current project - attachments uploaded as answers to 

56 questions with File Upload question elements 

57 """ 

58 project = fetch.project(session, project_id) 

59 validate.check( 

60 user, 

61 perms.ISSUE_VIEW_ANSWERS, 

62 project=project, 

63 deny_restricted=False, 

64 section_id=project.section_id, 

65 ) 

66 return [ 

67 serial.AnswerAttachment.model_validate(att) 

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

69 ] 

70 

71 

72@http 

73def get_project_attachment(session, user, project_id, attachment_id) -> XAccelResponse: 

74 """ 

75 Download the Project Attachment with the given ID 

76 """ 

77 project = fetch.project(session, project_id) 

78 validate.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=False) 

79 attachment = project.get_attachment(user, attachment_id) 

80 

81 return XAccelAttachmentResponse(attachment) 

82 

83 

84@http 

85def post_project_attachment( 

86 session, user, project_id, attachment_upload, attachment_description 

87) -> serial.Id: 

88 """ 

89 Upload a file as an attachment to the project. 

90 

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

92 """ 

93 

94 project = fetch.project(session, project_id) 

95 validate.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

96 attachment = attachments.save_project_attachment( 

97 session, project_id, user, attachment_upload, attachment_description 

98 ) 

99 session.flush() 

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

101 evt = AuditEvent.create( 

102 session, 

103 evt_types.PROJECT_ATTACHMENT_ADDED, 

104 project=project, 

105 user=user, 

106 object_id=attachment.id, 

107 change_list=changes, 

108 ) 

109 session.add(evt) 

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

111 

112 

113@http 

114def delete_project_attachment(session, user, project_id, attachment_id): 

115 """ 

116 Delete the Project Attachment with the given ID 

117 

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

119 """ 

120 project = fetch.project(session, project_id) 

121 validate.check(user, perms.PROJECT_EDIT_COSMETIC, project=project) 

122 attachment = project.get_attachment(user, attachment_id) 

123 filename = attachment.filename 

124 attachments.delete_project_attachment(session, attachment) 

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

126 evt = AuditEvent.create( 

127 session, 

128 evt_types.PROJECT_ATTACHMENT_REMOVED, 

129 project=project, 

130 user=user, 

131 object_id=attachment_id, 

132 change_list=changes, 

133 ) 

134 session.add(evt) 

135 

136 

137@http 

138def get_project_issue_attachment( 

139 session, user, project_id, issue_id, attachment_id 

140) -> XAccelResponse: 

141 """ 

142 Download the Issue Attachment with the given ID 

143 """ 

144 project = fetch.project(session, project_id) 

145 issue = fetch.issue(session, issue_id) 

146 validate.check( 

147 user, 

148 perms.ISSUE_VIEW_ANSWERS, 

149 issue=issue, 

150 project=project, 

151 section_id=project.section_id, 

152 ) 

153 attachment = issue.get_attachment(attachment_id) 

154 

155 return XAccelAttachmentResponse(attachment) 

156 

157 

158@http 

159def get_question_element_attachment( 

160 session, user, question_id, element_id 

161) -> XAccelResponse: 

162 """ 

163 Download the Question Attachment associated with the given Element ID 

164 """ 

165 q_instance = session.query(QuestionInstance).filter_by(id=question_id).one() 

166 q_element = q_instance.question_def.get_element(element_id) 

167 validate.check( 

168 user, 

169 perms.PROJECT_VIEW_QUESTIONNAIRE, 

170 project=q_instance.project, 

171 section_id=q_instance.section_id, 

172 ) 

173 return XAccelAttachmentResponse(q_element.attachment) 

174 

175 

176@http 

177def get_answer_attachment(session, user, answer_id) -> XAccelResponse: 

178 """ 

179 Download the Answer Attachment with the given ID 

180 """ 

181 answer = fetch.answer(session, answer_id) 

182 issue = answer.issue 

183 validate.check( 

184 user, 

185 perms.ISSUE_VIEW_ANSWERS, 

186 issue=issue, 

187 project=issue.project, 

188 section_id=answer.question_instance.section_id, 

189 ) 

190 

191 if not answer.attachment: 

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

193 

194 return XAccelAttachmentResponse(answer.attachment)