Coverage for rfpy/api/endpoints/answers.py: 100%
63 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
1"""
2Operations for fetching vendor's answers
3"""
5from collections import defaultdict
6from typing import List
8from sqlalchemy.orm import Session
10from rfpy.suxint import http
11from rfpy.model import Issue, QuestionInstance, Answer, User
12from rfpy.api import fetch, validate
13from rfpy.auth import perms
14from rfpy.web import serial
17# Reusable chunk of documentation for openapi spec documentation of endpoint functions
18lookup_docs = """
19 Results are provided as a nested mapping object allowing easy reference when
20 assigning answers to questions for specific Issues:
21 ```
22 {
23 // ID of the Issue
24 "923": {
26 // Element ID to Answer Value
27 "2342742": "Yes, we can deliver to Jupiter",
28 "7234232": "No, we cannot deliver to Baltimore"
30 }
31 }
32 ```
33 So to reference an answer for Issue 923 to Element 7234232, the notation is
34 data['923']['7234232']
36 (bear in mind that numerical IDs are delivered as Strings when used as Object keys in JSON)
37"""
40@http
41def get_question_answers(session, user, question_id) -> serial.AnswerLookup:
42 """
43 Fetch answers for question `question_id` from project structures as nested object
44 mapping issue ID to question element ID to answer.
45 """
46 # all standard users get the same copy
47 question = fetch.question(session, question_id)
48 validate.check(
49 user,
50 perms.ISSUE_VIEW_ANSWERS,
51 project=question.project,
52 section_id=question.section_id,
53 )
55 def get_answers():
56 qelements = defaultdict(dict)
57 for answer in question.all_answers():
58 qelements[str(answer.issue_id)][str(answer.element_id)] = answer.as_dict()
60 return qelements
62 return get_answers()
65get_question_answers.__doc__ += lookup_docs
68@http
69def get_project_issue_question(
70 session, user, project_id, issue_id, q_question_number
71) -> serial.SingleRespondentQuestion:
72 """
73 Get the answered question for the given project and issue.
74 """
75 project = fetch.project(session, project_id)
76 question = project.question_by_number(q_question_number)
77 issue = project.issue_by_id(issue_id)
78 validate.check(
79 user,
80 perms.ISSUE_VIEW_ANSWERS,
81 issue=issue,
82 project=project,
83 section_id=question.section_id,
84 )
85 return question.single_vendor_dict(issue)
88@http
89def get_project_section_issue_answers(
90 session, user, project_id, section_id, issue_id
91) -> List[serial.Answer]:
92 """
93 Fetch an array of Answer objects for the given Project, Issue and Section
94 """
95 project = fetch.project(session, project_id)
96 issue = fetch.issue(session, issue_id)
98 validate.check(
99 user,
100 perms.ISSUE_VIEW_ANSWERS,
101 project=project,
102 section_id=section_id,
103 issue=issue,
104 )
106 cols = (
107 Answer.element_id,
108 Answer.answer,
109 Answer.issue_id,
110 Answer.question_instance_id.label("question_id"),
111 )
112 answers = (
113 session.query(*cols)
114 .join(Issue)
115 .join(QuestionInstance)
116 .filter(
117 Issue.project == project,
118 Answer.issue == issue,
119 QuestionInstance.section_id == section_id,
120 )
121 )
123 return [a._asdict() for a in answers]
126@http
127def get_question_issue_answers(
128 session: Session, user: User, question_id: int, issue_id: int
129) -> serial.ElementAnswerList:
130 """
131 Fetch an array of Answer & Element ID for the given Question, Issue
132 """
133 question = fetch.question(session, question_id)
134 issue = question.project.get_issue(issue_id)
136 validate.check(
137 user,
138 perms.ISSUE_VIEW_ANSWERS,
139 project=question.project,
140 section_id=question.section_id,
141 issue=issue,
142 )
144 result = serial.ElementAnswerList(root=[])
145 answers = question.answers_for_issue(issue.id).all()
146 for answer in answers:
147 result.root.append(
148 serial.ElementAnswer(element_id=answer.element_id, answer=answer.answer)
149 )
151 return result
154@http
155def get_element_answers(session, user, element_id) -> serial.RespondentAnswers:
156 """
157 An array of answers from all projects and all issues for the provided element_id.
159 Project ID, Title and Date Published are provided to be used as a reference when reviewing
160 previous answers.
161 """
162 validate.check(user, perms.ISSUE_VIEW_ANSWERS, multiproject=True)
163 return serial.RespondentAnswers(
164 [
165 serial.RespondentAnswer.model_validate(a)
166 for a in fetch.element_answers(session, user, element_id)
167 ]
168 )
171@http
172def get_project_answers(session, user, project_id, issue_ids) -> serial.AnswerLookup:
173 """
174 Fetch Answers to all questions in the current project for the Issue IDs provided.
176 __N.B.__ This operation is inefficient for projects with very many questions or Issues. A
177 maximum of 2,000 element answers will be returned. An HTTP 400 error will be returned if more
178 than 2,000 answers are found. This total does not include unanswered Question Element / Issue
179 combinations. Operation ID get_project_qstats can be used to check how many
180 answerable elements are in the project.
181 """
182 project = fetch.project(session, project_id)
183 validate.check(
184 user, perms.ISSUE_VIEW_ANSWERS, project=project, deny_restricted=True
185 )
187 issue_id_set = {i.id for i in project.scoreable_issues} & issue_ids
189 aq = fetch.answers_in_issues_query(session, project_id, issue_id_set)
191 if aq.count() > 2000:
192 m = "More than 2,000 results found: submit fewer issueIds or use a different operation"
193 raise ValueError(m)
195 adict: dict[str, dict[str, str]] = defaultdict(dict)
196 for issue_id, element_id, answer in aq.with_entities(
197 Answer.issue_id, Answer.element_id, Answer.answer
198 ):
199 adict[str(issue_id)][str(element_id)] = answer
201 return serial.AnswerLookup.model_construct(adict)
204get_project_answers.__doc__ += lookup_docs