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