Coverage for postrfp/buyer/api/endpoints/answers.py: 100%

1""" 

2Operations for fetching vendor's answers 

3""" 

4 

5from collections import defaultdict 

6 

7from sqlalchemy.orm import Session 

8 

9from postrfp.model.questionnaire.answering import Answer 

10from postrfp.shared import fetch 

11from postrfp.shared.decorators import http 

12from postrfp.model import User, Issue, QuestionInstance 

13from postrfp.buyer.api import authorise 

14from postrfp.authorisation import perms 

15from postrfp.shared import serial 

16 

17 

18# Reusable chunk of documentation for openapi spec documentation of endpoint functions 

19lookup_docs = """ 

20 Results are provided as a nested mapping object allowing easy reference when 

21 assigning answers to questions for specific Issues: 

22 ``` 

23 { 

24 // ID of the Issue 

25 "923": { 

26 

27 // Element ID to Answer Value 

28 "2342742": "Yes, we can deliver to Jupiter", 

29 "7234232": "No, we cannot deliver to Baltimore" 

30 

31 } 

32 } 

33 ``` 

34 So to reference an answer for Issue 923 to Element 7234232, the notation is 

35 data['923']['7234232'] 

36 

37 (bear in mind that numerical IDs are delivered as Strings when used as Object keys in JSON) 

38""" 

39 

40 

41@http 

42def get_question_answers( 

43 session: Session, user: User, question_id: int 

44) -> serial.AnswerLookup: 

45 """ 

46 Fetch answers for question `question_id` from project structures as nested object 

47 mapping issue ID to question element ID to answer. 

48 """ 

49 # Fetch question (ideally preloading project via fetch.question) 

50 question = fetch.question(session, question_id, with_project=True) 

51 # Validation remains in the endpoint 

52 authorise.check( 

53 user, 

54 perms.ISSUE_VIEW_ANSWERS, 

55 project=question.project, 

56 section_id=question.section_id, 

57 ) 

58 

59 answer_data = fetch.visible_answers(session, question) 

60 return serial.AnswerLookup.model_construct(answer_data) 

61 

62 

63get_question_answers.__doc__ = (get_question_answers.__doc__ or "") + lookup_docs 

64 

65 

66@http 

67def get_project_issue_question( 

68 session: Session, user: User, project_id: int, issue_id: int, q_question_number: str 

69) -> serial.SingleRespondentQuestion: 

70 """ 

71 Get the answered question for the given project and issue. 

72 """ 

73 project = fetch.project(session, project_id) 

74 question = project.question_by_number(q_question_number) 

75 issue = project.issue_by_id(issue_id) 

76 authorise.check( 

77 user, 

78 perms.ISSUE_VIEW_ANSWERS, 

79 issue=issue, 

80 project=project, 

81 section_id=question.section_id, 

82 ) 

83 # TODO Fix schemas to permit answered questions. 

84 return question.single_vendor_dict(issue) # type: ignore[return-value] 

85 

86 

87@http 

88def get_project_section_issue_answers( 

89 session: Session, user: User, project_id: int, section_id: int, issue_id: int 

90) -> list[serial.Answer]: 

91 """ 

92 Fetch an array of Answer objects for the given Project, Issue and Section 

93 """ 

94 project = fetch.project(session, project_id) 

95 issue = fetch.issue(session, issue_id) 

96 

97 authorise.check( 

98 user, 

99 perms.ISSUE_VIEW_ANSWERS, 

100 project=project, 

101 section_id=section_id, 

102 issue=issue, 

103 ) 

104 

105 cols = ( 

106 Answer.element_id, 

107 Answer.answer, 

108 Answer.issue_id, 

109 Answer.question_instance_id.label("question_id"), 

110 ) 

111 answers = ( 

112 session.query(*cols) 

113 .join(Issue) 

114 .join(QuestionInstance) 

115 .filter( 

116 Issue.project == project, 

117 Answer.issue == issue, 

118 QuestionInstance.section_id == section_id, 

119 ) 

120 ) 

121 

122 return [serial.Answer(**a._asdict()) for a in answers] 

123 

124 

125@http 

126def get_question_issue_answers( 

127 session: Session, user: User, question_id: int, issue_id: int 

128) -> serial.ElementAnswerList: 

129 """ 

130 Fetch an array of Answer & Element ID for the given Question, Issue 

131 """ 

132 question = fetch.question(session, question_id, with_project=True) 

133 issue = question.project.get_issue(issue_id) 

134 

135 authorise.check( 

136 user, 

137 perms.ISSUE_VIEW_ANSWERS, 

138 project=question.project, 

139 section_id=question.section_id, 

140 issue=issue, 

141 ) 

142 

143 result = serial.ElementAnswerList(root=[]) 

144 answers = question.answers_for_issue(issue.id).all() 

145 for answer in answers: 

146 result.root.append( 

147 serial.ElementAnswer(element_id=answer.element_id, answer=answer.answer) 

148 ) 

149 

150 return result 

151 

152 

153@http 

154def get_element_answers( 

155 session: Session, user: User, element_id: int 

156) -> serial.RespondentAnswers: 

157 """ 

158 An array of answers from all projects and all issues for the provided element_id. 

159 

160 Project ID, Title and Date Published are provided to be used as a reference when reviewing 

161 previous answers. 

162 """ 

163 authorise.check(user, perms.ISSUE_VIEW_ANSWERS, multiproject=True) 

164 return serial.RespondentAnswers( 

165 [ 

166 serial.RespondentAnswer.model_validate(a) 

167 for a in fetch.element_answers(session, user, element_id) 

168 ] 

169 ) 

170 

171 

172@http 

173def get_project_answers( 

174 session: Session, user: User, project_id: int, issue_ids: set[int] 

175) -> serial.AnswerLookup: 

176 """ 

177 Fetch Answers to all questions in the current project for the Issue IDs provided. 

178 

179 __N.B.__ This operation is inefficient for projects with very many questions or Issues. A 

180 maximum of 2,000 element answers will be returned. An HTTP 400 error will be returned if more 

181 than 2,000 answers are found. This total does not include unanswered Question Element / Issue 

182 combinations. Operation ID get_project_qstats can be used to check how many 

183 answerable elements are in the project. 

184 """ 

185 project = fetch.project(session, project_id) 

186 authorise.check( 

187 user, perms.ISSUE_VIEW_ANSWERS, project=project, deny_restricted=True 

188 ) 

189 

190 issue_id_set = {i.id for i in project.scoreable_issues} & issue_ids 

191 

192 aq = fetch.answers_in_issues_query(session, project_id, issue_id_set) 

193 

194 if aq.count() > 2000: 

195 m = "More than 2,000 results found: submit fewer issueIds or use a different operation" 

196 raise ValueError(m) 

197 

198 adict: dict[str, dict[str, str]] = defaultdict(dict) 

199 for issue_id, element_id, answer in aq.with_entities( 

200 Answer.issue_id, Answer.element_id, Answer.answer 

201 ): 

202 adict[str(issue_id)][str(element_id)] = answer 

203 

204 return serial.AnswerLookup.model_construct(adict) 

205 

206 

207get_project_answers.__doc__ = (get_project_answers.__doc__ or "") + lookup_docs