Coverage for rfpy/api/endpoints/answers.py: 87%

1''' 

2Operations for fetching vendor's answers 

3''' 

4from collections import defaultdict 

5from typing import List 

6 

7from rfpy.suxint import http 

8from rfpy.model import ( 

9 Issue, QuestionInstance, Answer 

10) 

11from rfpy.api import fetch, validate 

12from rfpy.auth import perms 

13from rfpy.web import serial 

14 

15 

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

17lookup_docs = ''' 

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

19 assigning answers to questions for specific Issues: 

20 ``` 

21 { 

22 // ID of the Issue 

23 "923": { 

24 

25 // Element ID to Answer Value 

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

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

28 

29 } 

30 } 

31 ``` 

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

33 data['923']['7234232'] 

34 

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

36''' 

37 

38 

39@http 

40def get_question_answers(session, user, question_id) -> serial.AnswerLookup: 

41 """ 

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

43 mapping issue ID to question element ID to answer. 

44 """ 

45 # all standard users get the same copy 

46 question = fetch.question(session, question_id) 

47 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

48 project=question.project, 

49 section_id=question.section_id) 

50 

51 def get_answers(): 

52 qelements = defaultdict(dict) 

53 for answer in question.all_answers(): 

54 qelements[str(answer.issue_id)][str(answer.element_id)] = answer.as_dict() 

55 

56 return qelements 

57 

58 return get_answers() 

59 

60 

61get_question_answers.__doc__ += lookup_docs 

62 

63 

64@http 

65def get_project_issue_question(session, user, 

66 project_id, issue_id, 

67 q_question_number) -> serial.SingleRespondentQuestion: 

68 ''' 

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

70 ''' 

71 project = fetch.project(session, project_id) 

72 question = project.question_by_number(q_question_number) 

73 issue = project.issue_by_id(issue_id) 

74 validate.check(user, perms.ISSUE_VIEW_ANSWERS, issue=issue, 

75 project=project, section_id=question.section_id) 

76 return question.single_vendor_dict(issue) 

77 

78 

79@http 

80def get_project_section_issue_answers(session, 

81 user, 

82 project_id, 

83 section_id, 

84 issue_id) -> List[serial.Answer]: 

85 ''' 

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

87 ''' 

88 project = fetch.project(session, project_id) 

89 issue = fetch.issue(session, issue_id) 

90 

91 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

92 project=project, section_id=section_id, 

93 issue=issue) 

94 

95 cols = ( 

96 Answer.element_id, 

97 Answer.answer, 

98 Answer.issue_id, 

99 Answer.question_instance_id.label('question_id') 

100 ) 

101 answers = session.query(*cols)\ 

102 .join(Issue)\ 

103 .join(QuestionInstance)\ 

104 .filter( 

105 Issue.project == project, 

106 Answer.issue == issue, 

107 QuestionInstance.section_id == section_id) 

108 

109 return [a._asdict() for a in answers] 

110 

111@http 

112def get_question_issue_answers(session, user, question_id, issue_id) -> List[serial.ElementAnswerList]: 

113 ''' 

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

115 ''' 

116 question = fetch.question(session, question_id) 

117 issue = question.project.get_issue(issue_id) 

118 

119 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

120 project=question.project, section_id=question.section_id, 

121 issue=issue) 

122 

123 result = [] 

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

125 for answer in answers: 

126 result.append({ 

127 "element_id": answer.element_id, 

128 "answer": answer.answer 

129 }) 

130 

131 return result 

132 

133@http 

134def get_element_answers(session, user, element_id) -> serial.RespondentAnswers: 

135 ''' 

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

137 

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

139 previous answers. 

140 ''' 

141 validate.check(user, perms.ISSUE_VIEW_ANSWERS, multiproject=True) 

142 return [a._asdict() for a in fetch.element_answers(session, user, element_id)] 

143 

144 

145@http 

146def get_project_answers(session, user, project_id, issue_ids) -> serial.AnswerLookup: 

147 ''' 

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

149 

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

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

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

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

154 answerable elements are in the project. 

155 ''' 

156 project = fetch.project(session, project_id) 

157 validate.check(user, perms.ISSUE_VIEW_ANSWERS, 

158 project=project, deny_restricted=True) 

159 

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

161 

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

163 

164 if aq.count() > 2000: 

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

166 raise ValueError(m) 

167 

168 adict = defaultdict(dict) 

169 for issue_id, element_id, answer in aq.with_entities(Answer.issue_id, 

170 Answer.element_id, 

171 Answer.answer): 

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

173 

174 return adict 

175 

176 

177get_project_answers.__doc__ += lookup_docs