Coverage for rfpy/api/endpoints/sections.py: 86%
206 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-31 16:00 +0000
1'''
2Create, edit and view sections within a project
3'''
4from rfpy.model.helpers import validate_section_children
5from typing import List
6from rfpy.model.composite import update_qmeta_table
7from cgi import FieldStorage
9from sqlalchemy import or_, func
10from sqlalchemy.orm import Session
11from sqlalchemy.orm.exc import NoResultFound
13from rfpy.suxint import http, http_etag
14from rfpy.model import (
15 QuestionInstance, User, Section, AuditEvent, QuestionDefinition, ImportType
16)
17from rfpy.api import fetch, validate, update
18from rfpy.model.audit import evt_types
19from rfpy.model.exc import QuestionnaireStructureException, DuplicateQuestionDefinition
20from rfpy.auth import perms
21from rfpy.web import serial
24@http
25def put_project_section(session: Session, user: User, project_id: int,
26 section_id: int, edit_sec_doc: dict) -> serial.QuestionDef:
27 '''Update title or description for the given Section'''
28 section = fetch.section_by_id(session, section_id)
29 project = section.project
30 validate.check(user, perms.PROJECT_EDIT_COSMETIC, project=project, section_id=section.id)
32 evt = AuditEvent.create(evt_types.SECTION_UPDATED, project=project, user_id=user.id,
33 org_id=user.organisation.id, object_id=section.id, private=True)
35 for attr_name in ('description', 'title'):
36 if attr_name in edit_sec_doc:
37 new_val = edit_sec_doc[attr_name]
38 old_val = getattr(section, attr_name)
39 if new_val != old_val:
40 evt.add_change(attr_name, old_val, new_val)
41 setattr(section, attr_name, new_val)
43 session.add(evt)
46@http
47def post_project_section(session: Session, user: User, project_id: int,
48 section_id: int, edit_sec_doc: serial.Section) -> serial.Section:
49 '''Create a new Section'''
50 parent = fetch.section_by_id(session, section_id)
51 project = parent.project
52 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section_id)
54 new_section = Section(title=edit_sec_doc['title'],
55 description=edit_sec_doc.get('description', None))
56 parent.subsections.append(new_section)
57 session.flush()
58 evt = AuditEvent.create(evt_types.SECTION_CREATED, project=project, user_id=user.id,
59 org_id=user.organisation.id, object_id=new_section.id, private=True)
60 evt.add_change('title', None, new_section.title)
61 if edit_sec_doc.get('description', False):
62 evt.add_change('description', None, new_section.description)
63 session.add(evt)
64 return serial.Section.from_orm(new_section)
67@http
68def delete_project_section(session: Session, user: User, project_id: int, section_id: int):
69 '''
70 Delete the given Section and all questions and subsections contained within that
71 section.
73 @permission PROJECT_SAVE_QUESTIONNAIRE
74 '''
75 section = fetch.section_by_id(session, section_id)
76 project = section.project
78 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
79 project=project, section_id=section_id)
81 return update.delete_project_section(session, user, project, section)
84@http
85def put_project_movesection(session: Session, user: User, project_id: int,
86 move_section_doc: serial.MoveSection) -> serial.Section:
87 '''
88 Move the section given by section_id to a location with the same project given by
89 new_parent_id - the ID of the new Parent section
90 '''
91 project = fetch.project(session, project_id)
92 new_parent_id = move_section_doc['new_parent_id']
93 section_id = move_section_doc['section_id']
95 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
96 project=project, section_id=section_id,
97 deny_restricted=True)
98 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
99 project=project, section_id=new_parent_id,
100 deny_restricted=True)
102 section = project.sections.filter_by(id=section_id).one()
103 parent_section = project.sections.filter_by(id=new_parent_id).one()
105 if section is parent_section:
106 m = 'section_id and new_parent_id must be different values'
107 raise QuestionnaireStructureException(m)
109 if section.parent_id == new_parent_id:
110 raise ValueError('parent_section_id provided cannot be the same as the current parent_id')
112 section.parent_id = new_parent_id
113 session.flush()
114 if parent_section.parent:
115 parent_section.parent.renumber()
116 else:
117 parent_section.renumber()
118 session.refresh(section)
119 return serial.Section.from_orm(section)
122@http
123def put_project_section_children(
124 session: Session,
125 user: User,
126 project_id: int,
127 section_id: int,
128 child_nodes_doc: serial.SectionChildNodes) -> serial.SummarySection:
129 '''
130 Set the contents of the Section by providing a list of section IDs *or* question IDs.
132 This method can be used to move *existing* questions or sections to a new parent section or
133 to re-order existing section contents. It cannot be used to move questions or sections from a
134 different project.
136 Child objects (Section or Question) which exist within the current section but whose ID is not
137 included in the JSON body parameter are termed 'orphans'. If 'delete_orphans' is true, then
138 such objects will be deleted, i.e. the contents of the current section will be exactly the same
139 as that provided by the question_ids or section_ids parameter.
141 The default value for 'delete_orphans' is false - this value should ony be overridden with
142 care.
144 N.B. The JSON body object should contained either question_ids or section_ids arrays, but
145 not both.
146 '''
147 project = fetch.project(session, project_id)
148 parent: Section = project.sections.filter_by(id=section_id).one()
149 validate.check(user, perms.PROJECT_EDIT, project=project, section_id=parent.id)
151 section_ids, question_ids = child_nodes_doc['section_ids'], child_nodes_doc['question_ids']
153 if len(question_ids) > 0:
154 NodeType = QuestionInstance
155 parent_id_attr = 'section_id'
156 children = parent.questions
157 proj_collection = project.questions
158 provided_node_ids = question_ids
159 else: # schema validation must prevent both section_ids and question_ids existing
160 NodeType = Section
161 parent_id_attr = 'parent_id'
162 children = parent.subsections
163 proj_collection = project.sections
164 provided_node_ids = section_ids
166 nodes = proj_collection.filter(NodeType.id.in_(provided_node_ids)).all()
168 provided_id_set = set(provided_node_ids)
170 if NodeType is Section and section_id in provided_id_set:
171 m = f'Parent section ID {section_id} (from URL param) cannot be assigned under section_ids'
172 raise ValueError(m)
174 found_node_ids = {n.id for n in nodes}
175 current_child_ids = {c.id for c in children}
176 moved_node_ids = found_node_ids - current_child_ids
177 delete_orphans = child_nodes_doc.get('delete_orphans', False)
179 validate_section_children(session, NodeType, provided_id_set,
180 found_node_ids, current_child_ids,
181 delete_orphans=delete_orphans)
183 # Record the parent sections of nodes being moved so we can renumber at the end
184 ex_parents = {session.query(NodeType).get(moved_id).parent for moved_id in moved_node_ids}
186 if delete_orphans: # Double check -
187 # - if orphans exist but delete_orphans is true then validation should fail
188 orphans = current_child_ids - provided_id_set
189 for orphan_id in orphans:
190 orphan = session.query(NodeType).get(orphan_id)
191 if orphan in ex_parents:
192 ex_parents.remove(orphan) # Don't try to renumber something deleted
193 if NodeType is Section:
194 update.delete_project_section(session, user, project, orphan)
195 else:
196 update.delete_project_section_question(
197 session, user, project, orphan.section, orphan)
199 children.clear()
200 node_lookup = {n.id: n for n in nodes}
201 for idx, node_id in enumerate(
202 list(dict.fromkeys(provided_node_ids))): # iterate ordered list from json param
203 node = node_lookup[node_id]
204 node.position = idx
205 setattr(node, parent_id_attr, section_id)
206 children.append(node)
208 parent.renumber(parent.number.dotted)
210 # Run this last so ex-children have completed their move
211 for ex_parent in ex_parents:
212 ex_parent.renumber(ex_parent.number.dotted)
214 # TODO - Audit Events, especially for deleting
216 return get_project_section(session, user, project_id, section_id)
219@http
220def get_project_section(session: Session, user: User, project_id: int,
221 section_id: int, with_qdefs=False) -> serial.SummarySection:
222 '''
223 Get the Section with the given ID together with questions and subsections contained
224 within the Section.
225 '''
226 project = fetch.project(session, project_id)
227 section = fetch.section_of_project(project, section_id)
228 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE,
229 project=project, section_id=section.id, deny_restricted=False)
231 sec_dict = {
232 "id": section.id,
233 "parent_id": section.parent_id,
234 "number": section.safe_number,
235 "title": section.title,
236 "description": section.description,
237 "subsections": [],
238 "questions": []
239 }
241 if with_qdefs:
242 qlist = [q.as_dict() for q in section.questions]
243 else:
244 sq = section.questions_query\
245 .join(QuestionDefinition, QuestionInstance.question_def_id == QuestionDefinition.id)\
246 .with_entities(QuestionInstance.id, QuestionDefinition.title, QuestionInstance.number)
247 qlist = [dict(id=r.id, number=r.number.dotted, title=r.title) for r in sq]
249 sec_dict['questions'] = qlist
251 for subsec in fetch.visible_subsections_query(section, user):
252 sec_dict['subsections'].append(
253 {'id': subsec.id, 'title': subsec.title, 'number': subsec.safe_number}
254 )
256 return sec_dict
259@http_etag
260def get_project_treenodes(
261 session: Session,
262 user: User,
263 project_id: int) -> List[serial.ProjectNode]:
264 '''
265 N.B. - this method is deprecated. Use get_project_nodes instead.
267 Get an array of all sections, subsections and questions for the given project id.
269 Each section or question is a "node" in the questionnaire tree structure. The array
270 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
271 '''
272 project = fetch.project(session, project_id)
273 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project,
274 deny_restricted=True)
275 nodes = []
277 for node in fetch.light_nodes(session, project_id):
278 nodes.append({
279 'id': node.id,
280 'title': node.title,
281 'type': node.type,
282 'parent_id': node.parent_id,
283 'number': node.number.dotted,
284 'node_position': node.number,
285 'depth': int(node.depth)
286 })
287 return nodes
290@http
291def get_project_nodes(
292 session: Session,
293 user: User,
294 project_id: int,
295 q_section_id: int,
296 with_ancestors) -> List[serial.ProjectNode]:
297 '''
298 Get an array of subsections and questions for the given project id and section id.
300 If section ID is not provided then results for the root section of the project are returned.
302 If ancestors parameter is true or 1, then nodes of all direct ancestors and siblings of those
303 ancestors are returned, facilitating the respresentation of a tree structure.
305 Each section or question is a "node" in the questionnaire. The array
306 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
307 '''
308 project = fetch.project(session, project_id)
309 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project,
310 section_id=q_section_id, deny_restricted=True)
312 if q_section_id is None:
313 q_section_id = project.section_id
314 sec = session.query(Section).get(q_section_id)
315 if sec is None:
316 raise NoResultFound(f"Section with ID {q_section_id} not found")
318 nodes = []
319 for node in fetch.visible_nodes(session, sec, with_ancestors=with_ancestors):
320 nodes.append({
321 'id': node.id,
322 'title': node.title,
323 'description': node.description,
324 'type': node.type,
325 'parent_id': node.parent_id,
326 'number': node.number.dotted,
327 'node_position': node.number,
328 'depth': int(node.depth)
329 })
330 return nodes
333@http
334def post_project_section_excel(
335 session: Session,
336 user: User,
337 project_id: int,
338 section_id: int,
339 data_upload: FieldStorage) -> serial.ExcelImportResult:
340 '''
341 Create questions and subsections within the given section by uploading an Excel spreadsheet
342 with a specific format of 5 columns:
344 - A: Section Title. Add the newly created question to this section, creating the Section if
345 necessary
346 - B: Question Title. Short - for reference and navigation
347 - C: Question Text. The body of the question
348 - D: Multiple Choices. Multiple choice options seperated by semi colons. A numeric value
349 in angle brackets, e.g. <5> after the option text is interpreted as the Autoscore value
350 for that option
351 - E: Comments Field header. If provided in conjunction with Multiple choice options the
352 value of this column is used to create a subheader below the multiple choice field
353 and above a text field for comments or qualifications.
356 At most 500 rows are processed. Additional rows are disregarded. If column D is empty the
357 question will consist of a single text input field.
358 '''
360 from .. io import excel_import
362 project = fetch.project(session, project_id)
363 section = fetch.section_of_project(project, section_id)
364 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
365 project=project, section_id=section_id,
366 deny_restricted=True)
367 eqi = excel_import.ExcelQImporter(section)
368 eqi.read_questions_excel(data_upload.file)
369 if section.is_top_level:
370 section.renumber()
371 else:
372 section.parent.renumber()
374 descendant_qids = (
375 project.questions
376 .filter(QuestionInstance.number.startswith(section.number))
377 .with_entities(QuestionInstance.question_def_id)
378 )
379 update_qmeta_table(session, {r[0] for r in descendant_qids})
381 return serial.ExcelImportResult(imported_count=eqi.created_count)
384@http
385def get_project_qstats(session: Session, user: User, project_id: int) -> serial.QuestionnaireStats:
386 '''
387 Retrieve statistics for counts of sections, questions and question elements (by type)
388 for the given project ID.
389 '''
390 project = fetch.project(session, project_id)
391 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project)
392 return fetch.questionnaire_stats(session, project_id)
395@http
396def post_project_section_import(
397 session: Session, user: User, project_id: int, section_id: int,
398 import_section_doc: serial.SectionImportDoc) -> serial.SectionImportResult:
399 '''
400 Import questionnaire from previous project
401 '''
403 des_project = fetch.project(session, project_id)
404 des_section = fetch.section_of_project(des_project, section_id)
406 src_project = fetch.project(session, import_section_doc["project_id"])
408 validate.check(user, perms.PROJECT_SAVE_QUESTIONNAIRE,
409 project=des_project, section_id=des_section.id)
410 validate.check(user, action=perms.PROJECT_ACCESS, project=src_project)
412 src_sections = []
413 for section_id in import_section_doc['section_ids']:
414 src_sections.append(fetch.section_of_project(src_project, section_id))
416 src_questions = []
417 for question_id in import_section_doc['question_ids']:
418 src_questions.append(fetch.question_of_project(src_project, question_id))
420 import_type = ImportType.COPY if import_section_doc['clone'] else ImportType.SHARE
421 if import_type == ImportType.SHARE:
422 duplicated_qis: List[QuestionInstance] = fetch.duplicated_qdefs(
423 session, project_id, import_section_doc["project_id"], src_sections, src_questions)
424 if len(duplicated_qis) > 0:
425 duplicated_titles = [f'{q.safe_number}. {q.title}' for q in duplicated_qis]
426 raise DuplicateQuestionDefinition(
427 f"Duplicate questions definition(s) {duplicated_titles}")
429 imported_sections: List[Section] = []
430 imported_questions: List[QuestionInstance] = []
431 for src_sec in src_sections:
432 sections, questions = update.import_section(session, src_sec, des_section, import_type)
433 imported_sections += sections
434 imported_questions += questions
435 for src_que in src_questions:
436 imported_qi: QuestionInstance = update.import_q_instance(src_que, des_section, import_type)
437 imported_questions.append(imported_qi)
439 session.flush()
440 if import_type == ImportType.COPY:
441 update_qmeta_table(session, {qi.question_def_id for qi in imported_questions})
443 if des_section.is_top_level:
444 des_section.renumber()
445 else:
446 des_section.parent.renumber()
448 return serial.SectionImportResult(
449 section_count=len(imported_sections),
450 question_count=len(imported_questions))