Coverage for rfpy/api/endpoints/sections.py: 97%
228 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-29 13:25 +0000
1"""
2Create, edit and view sections within a project
3"""
5from rfpy.model.helpers import validate_section_children
6from typing import List, Annotated
7from rfpy.model.composite import update_qmeta_table
8from cgi import FieldStorage
10from sqlalchemy import select
11from sqlalchemy.orm import Session
13from rfpy.suxint import http, http_etag
14from rfpy.model import (
15 QuestionInstance,
16 User,
17 Section,
18 AuditEvent,
19 QuestionDefinition,
20 ImportType,
21 Project,
22)
23from rfpy.api import fetch, validate, update
24from rfpy.model.audit import evt_types
25from rfpy.model.exc import QuestionnaireStructureException, DuplicateQuestionDefinition
26from rfpy.auth import perms
27from rfpy.web import serial
30@http
31def put_project_section(
32 session: Session,
33 user: User,
34 project_id: int,
35 section_id: int,
36 edit_sec_doc: serial.EditableSection,
37):
38 """Update title or description for the given Section"""
39 section = fetch.section_by_id(session, section_id)
40 project = section.project
41 validate.check(
42 user, perms.PROJECT_EDIT_COSMETIC, project=project, section_id=section.id
43 )
45 evt = AuditEvent.create(
46 session,
47 evt_types.SECTION_UPDATED,
48 project=project,
49 user_id=user.id,
50 org_id=user.organisation.id,
51 object_id=section.id,
52 private=True,
53 )
55 for attr_name in ("description", "title"):
56 new_val = getattr(edit_sec_doc, attr_name, None)
57 if new_val is not None:
58 old_val = getattr(section, attr_name)
59 if new_val != old_val:
60 evt.add_change(attr_name, old_val, new_val)
61 setattr(section, attr_name, new_val)
63 session.add(evt)
64 return serial.Section.model_validate(section)
67@http
68def post_project_section(
69 session: Session,
70 user: User,
71 project_id: int,
72 section_id: int,
73 edit_sec_doc: serial.EditableSection,
74) -> serial.Section:
75 """Create a new Section"""
76 parent = fetch.section_by_id(session, section_id)
77 project = fetch.project(session, project_id)
78 validate.check(
79 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section_id
80 )
82 new_section = Section(
83 title=edit_sec_doc.title, description=edit_sec_doc.description
84 )
85 parent.subsections.append(new_section)
86 session.flush()
87 evt = AuditEvent.create(
88 session,
89 evt_types.SECTION_CREATED,
90 project=project,
91 user_id=user.id,
92 org_id=user.organisation.id,
93 object_id=new_section.id,
94 private=True,
95 )
96 evt.add_change("title", None, new_section.title)
97 if edit_sec_doc.description:
98 evt.add_change("description", None, new_section.description)
99 session.add(evt)
100 return serial.Section.model_validate(new_section)
103@http
104def delete_project_section(
105 session: Session, user: User, project_id: int, section_id: int
106):
107 """
108 Delete the given Section and all questions and subsections contained within that
109 section.
111 @permission PROJECT_SAVE_QUESTIONNAIRE
112 """
113 section = fetch.section_by_id(session, section_id)
114 project = section.project
116 validate.check(
117 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section_id
118 )
120 return update.delete_project_section(session, user, project, section)
123@http
124def put_project_movesection(
125 session: Session, user: User, project_id: int, move_section_doc: serial.MoveSection
126) -> serial.Section:
127 """
128 Move the section given by section_id to a location with the same project given by
129 new_parent_id - the ID of the new Parent section
130 """
131 project = fetch.project(session, project_id)
132 new_parent_id = move_section_doc.new_parent_id
133 section_id = move_section_doc.section_id
135 validate.check(
136 user,
137 perms.PROJECT_SAVE_QUESTIONNAIRE,
138 project=project,
139 section_id=section_id,
140 deny_restricted=True,
141 )
142 validate.check(
143 user,
144 perms.PROJECT_SAVE_QUESTIONNAIRE,
145 project=project,
146 section_id=new_parent_id,
147 deny_restricted=True,
148 )
150 section = project.sections.filter_by(id=section_id).one()
151 parent_section = project.sections.filter_by(id=new_parent_id).one()
153 if section is parent_section:
154 m = "section_id and new_parent_id must be different values"
155 raise QuestionnaireStructureException(m)
157 if section.parent_id == new_parent_id:
158 raise ValueError(
159 "parent_section_id provided cannot be the same as the current parent_id"
160 )
162 section.parent_id = new_parent_id
163 session.flush()
164 if parent_section.parent:
165 parent_section.parent.renumber()
166 else:
167 parent_section.renumber()
168 session.refresh(section)
169 return serial.Section.model_validate(section)
172@http
173def put_project_section_children(
174 session: Session,
175 user: User,
176 project_id: int,
177 section_id: int,
178 child_nodes_doc: serial.SectionChildNodes,
179) -> serial.SummarySection:
180 """
181 Set the contents of the Section by providing a list of section IDs *or* question IDs.
183 This method can be used to move *existing* questions or sections to a new parent section or
184 to re-order existing section contents. It cannot be used to move questions or sections from a
185 different project.
187 Child objects (Section or Question) which exist within the current section but whose ID is not
188 included in the JSON body parameter are termed 'orphans'. If 'delete_orphans' is true, then
189 such objects will be deleted, i.e. the contents of the current section will be exactly the same
190 as that provided by the question_ids or section_ids parameter.
192 The default value for 'delete_orphans' is false - this value should ony be overridden with
193 care.
195 N.B. The JSON body object should contained either question_ids or section_ids arrays, but
196 not both.
197 """
198 project = fetch.project(session, project_id)
199 parent: Section = project.sections.filter_by(id=section_id).one()
200 validate.check(user, perms.PROJECT_EDIT, project=project, section_id=parent.id)
202 section_ids = child_nodes_doc.section_ids
203 question_ids = child_nodes_doc.question_ids
205 if section_ids:
206 return update_section_children(
207 session, user, project, parent, section_ids, child_nodes_doc.delete_orphans
208 )
209 elif question_ids:
210 return update_question_children(
211 session, user, project, parent, question_ids, child_nodes_doc.delete_orphans
212 )
213 else:
214 raise ValueError("Either section_ids or question_ids must be provided")
217def update_section_children(
218 session: Session,
219 user: User,
220 project: Project,
221 parent: Section,
222 section_ids: List[int],
223 delete_orphans: bool,
224) -> serial.SummarySection:
225 nodes = project.sections.filter(Section.id.in_(section_ids)).all()
227 children = parent.subsections
229 provided_id_set = set(section_ids)
230 found_node_ids = {n.id for n in nodes}
231 current_child_ids = {c.id for c in children}
232 moved_node_ids = found_node_ids - current_child_ids
234 if parent.id in provided_id_set:
235 raise ValueError(
236 f"Parent section ID {parent.id} (from URL param) cannot be assigned under section_ids"
237 )
239 validate_section_children(
240 session,
241 Section,
242 provided_id_set,
243 found_node_ids,
244 current_child_ids,
245 delete_orphans=delete_orphans,
246 )
247 siq = select(Section).filter(Section.id.in_(moved_node_ids))
249 # current_parents are the parents of subsections being moved to the new parent
250 # from different locations in the questionnaire
251 current_parents = {s.parent for s in session.scalars(siq)}
253 if delete_orphans:
254 orphan_ids = current_child_ids - provided_id_set
256 for orphan_id in orphan_ids:
257 orphan_subsection = session.get(Section, orphan_id)
259 if orphan_subsection in current_parents:
260 current_parents.remove(orphan_subsection)
262 if orphan_subsection is not None:
263 update.delete_project_section(session, user, project, orphan_subsection)
265 children.clear()
266 node_lookup = {n.id: n for n in nodes}
267 for idx, node_id in enumerate(list(dict.fromkeys(section_ids))):
268 node = node_lookup[node_id]
269 node.position = idx
270 setattr(node, "parent_id", parent.id)
271 children.append(node)
273 parent.renumber(parent.number.dotted)
275 for ex_parent in current_parents:
276 if ex_parent is not None:
277 ex_parent.renumber(ex_parent.number.dotted)
279 return serial.SummarySection.model_validate(parent)
282def update_question_children(
283 session: Session,
284 user: User,
285 project: Project,
286 parent: Section,
287 question_ids: List[int],
288 delete_orphans: bool,
289) -> serial.SummarySection:
290 nodes = project.questions.filter(QuestionInstance.id.in_(question_ids)).all()
291 parent_id_attr = "section_id"
292 children = parent.questions
294 provided_id_set = set(question_ids)
295 found_node_ids = {n.id for n in nodes}
296 current_child_ids = {c.id for c in children}
297 moved_node_ids = found_node_ids - current_child_ids
299 validate_section_children(
300 session,
301 QuestionInstance,
302 provided_id_set,
303 found_node_ids,
304 current_child_ids,
305 delete_orphans=delete_orphans,
306 )
308 qiq = select(QuestionInstance).filter(QuestionInstance.id.in_(moved_node_ids))
309 ex_parents: set[Section | None] = {
310 qi.parent for qi in session.scalars(qiq).unique()
311 }
313 if delete_orphans:
314 orphans = current_child_ids - provided_id_set
315 for orphan_id in orphans:
316 orphan = session.get(QuestionInstance, orphan_id)
317 # if orphan in ex_parents:
318 # ex_parents.remove(orphan)
319 if orphan is not None:
320 update.delete_project_section_question(
321 session, user, project, orphan.section, orphan
322 )
324 children.clear()
325 node_lookup = {n.id: n for n in nodes}
326 for idx, node_id in enumerate(list(dict.fromkeys(question_ids))):
327 node = node_lookup[node_id]
328 node.position = idx
329 setattr(node, parent_id_attr, parent.id)
330 children.append(node)
332 parent.renumber(parent.number.dotted)
334 for ex_parent in ex_parents:
335 if ex_parent is not None:
336 ex_parent.renumber(ex_parent.number.dotted)
338 return serial.SummarySection.model_validate(parent)
341@http
342def get_project_section(
343 session: Session, user: User, project_id: int, section_id: int
344) -> serial.SummarySection:
345 """
346 Get the Section with the given ID together with questions and subsections contained
347 within the Section.
348 """
349 project = fetch.project(session, project_id)
350 section = fetch.section_of_project(project, section_id)
351 validate.check(
352 user,
353 perms.PROJECT_VIEW_QUESTIONNAIRE,
354 project=project,
355 section_id=section.id,
356 deny_restricted=False,
357 )
359 sec_doc = serial.SummarySection(
360 id=section.id,
361 parent_id=section.parent_id,
362 number=section.safe_number,
363 title=section.title,
364 description=section.description,
365 )
367 sq = section.questions_query.join(
368 QuestionDefinition,
369 QuestionInstance.question_def_id == QuestionDefinition.id,
370 ).with_entities(
371 QuestionInstance.id, QuestionDefinition.title, QuestionInstance.number
372 )
373 sec_doc.questions = [
374 serial.Node(id=r.id, title=r.title, number=r.number.dotted) for r in sq
375 ]
377 for subsec in fetch.visible_subsections_query(section, user):
378 sec_doc.subsections.append(
379 serial.Node(id=subsec.id, title=subsec.title, number=subsec.safe_number)
380 )
382 return sec_doc
385@http_etag
386def get_project_treenodes(
387 session: Session, user: User, project_id: int
388) -> List[serial.ProjectNode]:
389 """
390 N.B. - this method is deprecated. Use get_project_nodes instead.
392 Get an array of all sections, subsections and questions for the given project id.
394 Each section or question is a "node" in the questionnaire tree structure. The array
395 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
396 """
397 project = fetch.project(session, project_id)
398 validate.check(
399 user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project, deny_restricted=True
400 )
401 nodes = []
403 for node in fetch.light_nodes(session, project_id):
404 nodes.append(
405 serial.ProjectNode(
406 id=node.id,
407 title=node.title,
408 type=node.type,
409 parent_id=node.parent_id,
410 number=node.number.dotted,
411 node_position=node.number,
412 depth=int(node.depth),
413 )
414 )
415 return nodes
418@http
419def get_project_nodes(
420 session: Session, user: User, project_id: int, q_section_id: int, with_ancestors
421) -> List[serial.ProjectNode]:
422 """
423 Get an array of subsections and questions for the given project id and section id.
424 If section ID is not provided then results for the root section of the project are returned.
426 If ancestors parameter is true or 1, then nodes of all direct ancestors and siblings of those
427 ancestors are returned, facilitating the respresentation of a tree structure.
429 Each section or question is a "node" in the questionnaire. The array
430 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
431 """
432 project = fetch.project(session, project_id)
433 validate.check(
434 user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project, deny_restricted=True
435 )
437 if q_section_id is None:
438 q_section_id = project.section_id
439 sec = (
440 session.query(Section)
441 .filter(Section.id == q_section_id)
442 .filter(Section.project_id == project_id)
443 .one()
444 )
446 nodes = []
447 for node in fetch.visible_nodes(session, sec, with_ancestors=with_ancestors):
448 nodes.append(
449 serial.ProjectNode(
450 id=node.id,
451 title=node.title,
452 type=node.type,
453 parent_id=node.parent_id,
454 number=node.number.dotted,
455 node_position=node.number,
456 depth=int(node.depth),
457 )
458 )
459 return nodes
462@http
463def post_project_section_excel(
464 session: Session,
465 user: User,
466 project_id: int,
467 section_id: int,
468 data_upload: FieldStorage,
469) -> serial.ExcelImportResult:
470 """
471 Create questions and subsections within the given section by uploading an Excel spreadsheet
472 with a specific format of 5 columns:
474 - A: Section Title. Add the newly created question to this section, creating the Section if
475 necessary
476 - B: Question Title. Short - for reference and navigation
477 - C: Question Text. The body of the question
478 - D: Multiple Choices. Multiple choice options seperated by semi colons. A numeric value
479 in angle brackets, e.g. <5> after the option text is interpreted as the Autoscore value
480 for that option
481 - E: Comments Field header. If provided in conjunction with Multiple choice options the
482 value of this column is used to create a subheader below the multiple choice field
483 and above a text field for comments or qualifications.
486 At most 500 rows are processed. Additional rows are disregarded. If column D is empty the
487 question will consist of a single text input field.
488 """
490 from ..io import excel_import
492 project = fetch.project(session, project_id)
493 section = fetch.section_of_project(project, section_id)
494 validate.check(
495 user,
496 perms.PROJECT_SAVE_QUESTIONNAIRE,
497 project=project,
498 section_id=section_id,
499 deny_restricted=True,
500 )
501 eqi = excel_import.ExcelQImporter(section)
502 eqi.read_questions_excel(data_upload.file)
503 if section.is_top_level:
504 section.renumber()
505 elif section.parent is not None:
506 section.parent.renumber()
508 descendant_qids = project.questions.filter(
509 QuestionInstance.number.startswith(section.number)
510 ).with_entities(QuestionInstance.question_def_id)
511 update_qmeta_table(session, {r[0] for r in descendant_qids})
513 return serial.ExcelImportResult(imported_count=eqi.created_count)
516@http
517def get_project_qstats(
518 session: Session, user: User, project_id: int
519) -> Annotated[dict[str, int | dict], serial.QuestionnaireStats]:
520 """
521 Retrieve statistics for counts of sections, questions and question elements (by type)
522 for the given project ID.
523 """
524 project = fetch.project(session, project_id)
525 validate.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project)
526 return fetch.questionnaire_stats(session, project_id)
529@http
530def post_project_section_import(
531 session: Session,
532 user: User,
533 project_id: int,
534 section_id: int,
535 import_section_doc: serial.SectionImportDoc,
536) -> serial.SectionImportResult:
537 """
538 Import questionnaire from previous project
539 """
541 des_project = fetch.project(session, project_id)
542 des_section = fetch.section_of_project(des_project, section_id)
544 src_project = fetch.project(session, import_section_doc.project_id)
546 validate.check(
547 user,
548 perms.PROJECT_SAVE_QUESTIONNAIRE,
549 project=des_project,
550 section_id=des_section.id,
551 )
552 validate.check(user, action=perms.PROJECT_ACCESS, project=src_project)
554 src_sections = []
555 for section_id in import_section_doc.section_ids:
556 src_sections.append(fetch.section_of_project(src_project, section_id))
558 src_questions = []
559 for question_id in import_section_doc.question_ids:
560 src_questions.append(fetch.question_of_project(src_project, question_id))
562 import_type = ImportType.COPY if import_section_doc.clone else ImportType.SHARE
563 if import_type == ImportType.SHARE:
564 duplicated_qis: List[QuestionInstance] = fetch.duplicated_qdefs(
565 session,
566 project_id,
567 import_section_doc.project_id,
568 src_sections,
569 src_questions,
570 )
571 if len(duplicated_qis) > 0:
572 duplicated_titles = [f"{q.safe_number}. {q.title}" for q in duplicated_qis]
573 raise DuplicateQuestionDefinition(
574 f"Duplicate questions definition(s) {duplicated_titles}"
575 )
577 imported_sections: List[Section] = []
578 imported_questions: List[QuestionInstance] = []
579 for src_sec in src_sections:
580 sections, questions = update.import_section(
581 session, src_sec, des_section, import_type
582 )
583 imported_sections += sections
584 imported_questions += questions
585 for src_que in src_questions:
586 imported_qi: QuestionInstance = update.import_q_instance(
587 src_que, des_section, import_type
588 )
589 imported_questions.append(imported_qi)
591 session.flush()
592 if import_type == ImportType.COPY:
593 update_qmeta_table(session, {qi.question_def_id for qi in imported_questions})
595 if des_section.is_top_level:
596 des_section.renumber()
597 elif des_section.parent is not None:
598 des_section.parent.renumber()
600 return serial.SectionImportResult(
601 section_count=len(imported_sections), question_count=len(imported_questions)
602 )