Coverage for rfpy/api/endpoints/questions.py: 100%
123 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-31 16:00 +0000
2'''
3Create, edit and fetch questions within a project
4'''
5from rfpy.model.composite import update_qmeta_table
6from typing import Dict, List
8from sqlalchemy.orm import Session, subqueryload
10from rfpy.suxint import http
11from rfpy.model import (
12 QuestionInstance, User, AuditEvent, QuestionDefinition, Project, Participant, QElement
13)
14from rfpy.api import fetch, validate, update
15from rfpy.web import serial
16from rfpy.auth import perms
17from rfpy.model.audit import evt_types
20@http
21def get_project_section_question(session, user,
22 project_id: int,
23 section_id: int,
24 question_id) -> serial.Question:
25 '''Get the question with the given ID'''
26 qi = session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
27 project = qi.project
28 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE,
29 project=project,
30 section_id=qi.section_id)
31 return serial.Question.from_orm(qi)
34@http
35def put_project_section_question(session: Session,
36 user: User,
37 project_id: int,
38 section_id: int,
39 question_id: int,
40 qdef_doc: serial.QuestionDef) -> serial.Question:
41 '''
42 Update the title and question elements for the given question
44 The provided JSON document must provide elements with ID values for all elements that are to
45 be retained.
46 Elements with ID values provided are updated.
47 Elements without an ID are added as new.
48 Any existing elements not provided (with matching IDs) in the JSON document are deleted.
50 If items to be deleted have already been answered (in any project) this update is rejected
51 with an HTTP Status code of 409 (Conflict).
52 '''
54 qinstance: QuestionInstance = (session.query(QuestionInstance)
55 .filter(QuestionInstance.id == question_id)
56 .one())
57 project = qinstance.project
59 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
60 project=project, section_id=qinstance.section_id,
61 deny_restricted=True)
63 qdef: QuestionDefinition = qinstance.question_def
65 evt: AuditEvent = AuditEvent.create('QUESTION_EDITED',
66 project=project, user_id=user.id,
67 org_id=user.org_id, object_id=qinstance.id,
68 question_id=qinstance.id, private=True)
69 new_title = qdef_doc['title']
70 if qdef.title != new_title:
71 evt.add_change('Title', qdef.title, new_title)
72 qdef.title = new_title
74 el_map = {el.id: el for el in qdef.elements}
76 for row_number, row in enumerate(qdef_doc['elements'], start=1):
77 for col_number, el_dict in enumerate(row, start=1):
78 el_dict['row'] = row_number
79 el_dict['col'] = col_number
80 update.update_create_qdef(qdef, evt, el_map, el_dict)
82 update.check_for_saved_answers(session, qdef, el_map)
84 for _id, unwanted_element in el_map.items():
85 qdef.elements.remove(unwanted_element)
86 evt.add_change('Element Removed', unwanted_element.el_type, None)
88 session.flush()
89 update_qmeta_table(session, {qdef.id})
90 session.add(evt)
91 qdef.elements.sort(key=lambda el: (el.row, el.col))
92 return serial.Question.from_orm(qinstance).dict()
95@http
96def post_project_section_question(session,
97 user,
98 project_id,
99 section_id,
100 qdef_doc: serial.QuestionDef) -> serial.Question:
101 '''Create a new Question in the given section'''
102 section = fetch.section_by_id(session, section_id)
103 project = section.project
104 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
105 project=project, section_id=section.id, deny_restricted=True)
107 qi = QuestionInstance(project=project)
108 qi.question_def = QuestionDefinition.build(qdef_doc, strip_ids=True)
109 section.questions.append(qi)
110 session.flush()
111 evt = AuditEvent.create('QUESTION_CREATED',
112 project=project,
113 user_id=user.id,
114 org_id=user.organisation.id,
115 object_id=qi.id,
116 private=True,
117 question_id=qi.id)
118 evt.add_change('Title', None, qi.title)
120 update_qmeta_table(session, {qi.question_def_id})
121 session.add(evt)
122 return serial.Question.from_orm(qi).dict()
125@http
126def delete_project_section_question(session,
127 user,
128 project_id,
129 section_id,
130 question_id) -> List[serial.QI]:
131 '''
132 Delete the Question with the given ID
134 The return value is an array of remaining instances of the same question that may exist
135 in other projects
137 @permission PROJECT_SAVE_QUESTIONNAIRE
138 '''
140 qi: QuestionInstance = (session.query(QuestionInstance)
141 .filter(QuestionInstance.id == question_id)
142 .one())
143 if qi.section_id != section_id:
144 raise ValueError(f'Question #{question_id} does not belong to section #{section_id}')
145 project = qi.project
146 section = qi.section
148 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
149 project=project, section_id=section_id,
150 deny_restricted=True)
152 return update.delete_project_section_question(session, user, project, section, qi)
155@http
156def post_project_section_question_copy(session: Session, user: User, project_id: int,
157 section_id: int, question_id: int) -> serial.Node:
158 '''Create a copy of the question (instance & definition) given by question_id'''
160 section = fetch.section_by_id(session, section_id)
161 project = section.project
163 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section.id)
165 qi = fetch.question_of_section(session, section_id, question_id)
167 original_qdef = qi.question_def
168 copied_qdef = update.copy_q_definition(original_qdef, session)
170 new_qi: QuestionInstance = QuestionInstance(question_def=copied_qdef)
171 copied_qdef.title += f' (copied from {qi.number.dotted})'
172 section.questions.append(new_qi)
174 session.flush()
176 evt = AuditEvent.create('QUESTION_COPIED',
177 project=project, user_id=user.id,
178 org_id=user.organisation.id, object_id=qi.id,
179 private=True, question_id=new_qi.id)
181 evt.add_change('parent_id', None, original_qdef.id)
182 evt.add_change('title', original_qdef.title, copied_qdef.title)
183 evt.add_change('question_id', qi.id, new_qi.id)
184 evt.add_change('question_number', qi.number.dotted, new_qi.number.dotted)
185 session.add(evt)
187 update_qmeta_table(session, {copied_qdef.id})
189 return serial.Node(id=new_qi.id, number=new_qi.number.dotted,
190 title=new_qi.question_def.title)
193@http
194def get_question_instances(session: Session,
195 user: User,
196 question_id: int) -> List[serial.QuestionInstance]:
197 '''Find shared instances of the given question across all projects'''
198 qi = session.query(QuestionInstance).filter(QuestionInstance.id == question_id).one()
199 project = qi.project
200 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE,
201 project=project,
202 section_id=qi.section_id)
204 instances_query = qi.question_def.instances\
205 .join(Project, Participant)\
206 .filter(Participant.organisation == user.organisation)\
207 .options(subqueryload('project'))
209 instances = [{'project_id': inst.project.id,
210 'project_title': inst.project.title,
211 'number': inst.number.dotted,
212 'id': inst.id,
213 'section_id': inst.section_id}
214 for inst in instances_query.all()]
215 return instances
218@http
219def put_project_question_element(session: Session, user: User,
220 project_id: int, question_id: int,
221 element_id: int, element_doc: serial.QElement):
222 '''Update a single Question Element'''
223 project = fetch.project(session, project_id)
224 qinstance: QuestionInstance = project.questions.filter(QuestionInstance.id == question_id).one()
225 element: QElement = qinstance.question_def.get_element(element_id)
226 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
227 project=project, section_id=qinstance.section_id,
228 deny_restricted=True)
230 evt: AuditEvent = AuditEvent.create('QUESTION_EDITED',
231 project=project, user_id=user.id,
232 org_id=user.org_id, object_id=qinstance.id,
233 question_id=qinstance.id, private=True)
234 for k, v in element_doc.items():
235 if hasattr(element, k):
236 evt.add_change(k, getattr(element, k, v), v)
237 setattr(element, k, v)
239 session.add(evt)
240 update_qmeta_table(session, {question_id})
243@http
244def post_project_section_question_unlink(session: Session,
245 user: User,
246 project_id,
247 section_id: int,
248 question_id: int):
249 '''
250 Unlink the question from any previous projects so it can be edited freely.
251 N.B. This breaks answer & score importing between projects
252 '''
253 '''Create a copy of the question given by question_id'''
254 section = fetch.section_by_id(session, section_id)
255 project = section.project
257 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section.id)
258 qi: QuestionInstance = fetch.question_of_section(session, section_id, question_id)
259 original_qdef = qi.question_def
260 original_qdef.refcount = original_qdef.refcount - 1
262 copied_qdef = update.copy_q_definition(original_qdef, session)
263 qi.question_def = copied_qdef
264 qi.question_def.title += ' (copy unlinked from previous projects)'
266 evt = AuditEvent.create(evt_types.QUESTION_COPIED, project=project, user_id=user.id,
267 org_id=user.organisation.id, object_id=qi.id,
268 private=True, question_id=qi.id)
269 session.flush()
270 evt.add_change('parent_id', None, copied_qdef.parent_id)
271 evt.add_change('title', original_qdef.title, copied_qdef.title)
272 session.add(evt)