Coverage for rfpy/api/endpoints/questions.py: 100%
125 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
1"""
2Create, edit and fetch questions within a project
3"""
5from rfpy.model.composite import update_qmeta_table
6from typing import List
8from sqlalchemy.orm import Session, subqueryload
10from rfpy.suxint import http
11from rfpy.model import (
12 QuestionInstance,
13 User,
14 AuditEvent,
15 QuestionDefinition,
16 Project,
17 Participant,
18 QElement,
19)
20from rfpy.api import fetch, validate, update
21from rfpy.web import serial
22from rfpy.auth import perms
23from rfpy.model.audit import evt_types
26@http
27def get_project_section_question(
28 session, user, project_id: int, section_id: int, question_id
29) -> serial.Question:
30 """Get the question with the given ID"""
31 qi = (
32 session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
33 )
34 project = qi.project
35 validate.check(
36 user,
37 perms.PROJECT_VIEW_QUESTIONNAIRE,
38 project=project,
39 section_id=qi.section_id,
40 )
41 return serial.Question.model_validate(qi)
44@http
45def put_project_section_question(
46 session: Session,
47 user: User,
48 project_id: int,
49 section_id: int,
50 question_id: int,
51 qdef_doc: serial.QuestionDef,
52) -> serial.Question:
53 """
54 Update the title and question elements for the given question
56 The provided JSON document must provide elements with ID values for all elements that are to
57 be retained.
58 Elements with ID values provided are updated.
59 Elements without an ID are added as new.
60 Any existing elements not provided (with matching IDs) in the JSON document are deleted.
62 If items to be deleted have already been answered (in any project) this update is rejected
63 with an HTTP Status code of 409 (Conflict).
64 """
66 qinstance: QuestionInstance = (
67 session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
68 )
70 project = qinstance.project
72 validate.check(
73 user,
74 perms.PROJECT_SAVE_QUESTIONNAIRE,
75 project=project,
76 section_id=qinstance.section_id,
77 deny_restricted=True,
78 )
80 qdef: QuestionDefinition = qinstance.question_def
82 evt: AuditEvent = AuditEvent.create(
83 session,
84 "QUESTION_EDITED",
85 project=project,
86 user_id=user.id,
87 org_id=user.org_id,
88 object_id=qinstance.id,
89 question_id=qinstance.id,
90 private=True,
91 )
92 new_title = qdef_doc.title
93 if qdef.title != new_title:
94 evt.add_change("Title", qdef.title, new_title)
95 qdef.title = new_title
97 el_map = {el.id: el for el in qdef.elements}
99 for row_number, row in enumerate(qdef_doc.elements.root, start=1):
100 for col_number, el in enumerate(row.root, start=1):
101 el_dict = el.model_dump()
102 el_dict["row"] = row_number
103 el_dict["col"] = col_number
104 update.update_create_qdef(qdef, evt, el_map, el_dict)
106 update.check_for_saved_answers(session, qdef, el_map)
108 for _id, unwanted_element in el_map.items():
109 qdef.elements.remove(unwanted_element)
110 evt.add_change("Element Removed", unwanted_element.el_type, None)
112 session.flush()
113 update_qmeta_table(session, {qdef.id})
114 session.add(evt)
115 qdef.elements.sort(key=lambda el: (el.row, el.col))
116 return serial.Question.model_validate(qinstance)
119@http
120def post_project_section_question(
121 session, user, project_id, section_id, qdef_doc: serial.QuestionDef
122) -> serial.Question:
123 """Create a new Question in the given section"""
124 section = fetch.section_by_id(session, section_id)
125 project = section.project
126 validate.check(
127 user,
128 perms.PROJECT_SAVE_QUESTIONNAIRE,
129 project=project,
130 section_id=section.id,
131 deny_restricted=True,
132 )
134 qi = QuestionInstance(project=project)
135 qi.question_def = QuestionDefinition.build(qdef_doc, strip_ids=True)
136 section.questions.append(qi)
137 session.flush()
138 evt = AuditEvent.create(
139 session,
140 "QUESTION_CREATED",
141 project=project,
142 user_id=user.id,
143 org_id=user.organisation.id,
144 object_id=qi.id,
145 private=True,
146 question_id=qi.id,
147 )
148 evt.add_change("Title", None, qi.title)
150 update_qmeta_table(session, {qi.question_def_id})
151 return serial.Question.model_validate(qi)
154@http
155def delete_project_section_question(
156 session, user, project_id, section_id, question_id
157) -> List[serial.QI]:
158 """
159 Delete the Question with the given ID
161 The return value is an array of remaining instances of the same question that may exist
162 in other projects
164 @permission PROJECT_SAVE_QUESTIONNAIRE
165 """
167 qi: QuestionInstance = (
168 session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
169 )
170 if qi.section_id != section_id:
171 raise ValueError(
172 f"Question #{question_id} does not belong to section #{section_id}"
173 )
174 project = qi.project
175 section = qi.section
177 validate.check(
178 user,
179 perms.PROJECT_SAVE_QUESTIONNAIRE,
180 project=project,
181 section_id=section_id,
182 deny_restricted=True,
183 )
185 return update.delete_project_section_question(session, user, project, section, qi)
188@http
189def post_project_section_question_copy(
190 session: Session, user: User, project_id: int, section_id: int, question_id: int
191) -> serial.Node:
192 """Create a copy of the question (instance & definition) given by question_id"""
194 section = fetch.section_by_id(session, section_id)
195 project = section.project
197 validate.check(
198 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section.id
199 )
201 qi = fetch.question_of_section(session, section_id, question_id)
203 original_qdef = qi.question_def
204 copied_qdef = update.copy_q_definition(original_qdef, session)
206 new_qi: QuestionInstance = QuestionInstance(question_def=copied_qdef)
207 copied_qdef.title += f" (copied from {qi.number.dotted})"
208 section.questions.append(new_qi)
210 session.flush()
212 evt = AuditEvent.create(
213 session,
214 "QUESTION_COPIED",
215 project=project,
216 user_id=user.id,
217 org_id=user.organisation.id,
218 object_id=qi.id,
219 private=True,
220 question_id=new_qi.id,
221 )
223 evt.add_change("parent_id", None, original_qdef.id)
224 evt.add_change("title", original_qdef.title, copied_qdef.title)
225 evt.add_change("question_id", qi.id, new_qi.id)
226 evt.add_change("question_number", qi.number.dotted, new_qi.number.dotted)
227 session.add(evt)
229 update_qmeta_table(session, {copied_qdef.id})
231 return serial.Node(
232 id=new_qi.id, number=new_qi.number.dotted, title=new_qi.question_def.title
233 )
236@http
237def get_question_instances(
238 session: Session, user: User, question_id: int
239) -> List[serial.QuestionInstance]:
240 """Find shared instances of the given question across all projects"""
241 qi = (
242 session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
243 )
244 project = qi.project
245 validate.check(
246 user,
247 perms.PROJECT_VIEW_QUESTIONNAIRE,
248 project=project,
249 section_id=qi.section_id,
250 )
252 instances_query = (
253 qi.question_def.instances.join(Project)
254 .join(Participant)
255 .filter(Participant.organisation == user.organisation)
256 .options(subqueryload(QuestionInstance.project))
257 )
259 instances = [
260 serial.QuestionInstance.model_validate(inst) for inst in instances_query
261 ]
262 return instances
265@http
266def put_project_question_element(
267 session: Session,
268 user: User,
269 project_id: int,
270 question_id: int,
271 element_id: int,
272 element_doc: serial.QElement,
273):
274 """Update a single Question Element"""
275 project = fetch.project(session, project_id)
276 qinstance: QuestionInstance = project.questions.filter(
277 QuestionInstance.id == question_id
278 ).one()
279 element: QElement = qinstance.question_def.get_element(element_id)
280 validate.check(
281 user,
282 perms.PROJECT_SAVE_QUESTIONNAIRE,
283 project=project,
284 section_id=qinstance.section_id,
285 deny_restricted=True,
286 )
288 evt: AuditEvent = AuditEvent.create(
289 session,
290 "QUESTION_EDITED",
291 project=project,
292 user_id=user.id,
293 org_id=user.org_id,
294 object_id=qinstance.id,
295 question_id=qinstance.id,
296 private=True,
297 )
298 for doc_field_name in serial.QElement.model_fields.keys():
299 doc_field_value = getattr(element_doc, doc_field_name)
300 if hasattr(element, doc_field_name):
301 element_field_value = getattr(element, doc_field_name)
302 if element_field_value != doc_field_value:
303 evt.add_change(
304 doc_field_name, getattr(element, doc_field_name), doc_field_value
305 )
306 setattr(element, doc_field_name, doc_field_value)
308 session.add(evt)
309 update_qmeta_table(session, {question_id})
312@http
313def post_project_section_question_unlink(
314 session: Session, user: User, project_id, section_id: int, question_id: int
315):
316 """
317 Unlink the question from any previous projects so it can be edited freely.
318 N.B. This breaks answer & score importing between projects
319 """
320 """Create a copy of the question given by question_id"""
321 section = fetch.section_by_id(session, section_id)
322 project = section.project
324 validate.check(
325 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section.id
326 )
327 qi: QuestionInstance = fetch.question_of_section(session, section_id, question_id)
328 original_qdef = qi.question_def
329 original_qdef.refcount = original_qdef.refcount - 1
331 copied_qdef = update.copy_q_definition(original_qdef, session)
332 qi.question_def = copied_qdef
333 qi.question_def.title += " (copy unlinked from previous projects)"
335 evt = AuditEvent.create(
336 session,
337 evt_types.QUESTION_COPIED,
338 project=project,
339 user_id=user.id,
340 org_id=user.organisation.id,
341 object_id=qi.id,
342 private=True,
343 question_id=qi.id,
344 )
345 session.flush()
346 evt.add_change("parent_id", None, copied_qdef.parent_id)
347 evt.add_change("title", original_qdef.title, copied_qdef.title)
348 session.add(evt)