Coverage for postrfp/buyer/api/endpoints/sections.py: 97%
171 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-22 21:34 +0000
1"""
2Create, edit and view sections within a project
3"""
5from typing import List, Annotated
6from postrfp.model.composite import update_qmeta_table
7from cgi import FieldStorage
9from sqlalchemy.orm import Session
11from postrfp.shared import fetch, update, serial
12from postrfp.shared.decorators import http, http_etag
13from postrfp.model import (
14 QuestionInstance,
15 User,
16 Section,
17 AuditEvent,
18 QuestionDefinition,
19 ImportType,
20)
22from postrfp.buyer.api import authorise
23from postrfp.model.audit import evt_types
24from postrfp.model.exc import (
25 QuestionnaireStructureException,
26 DuplicateQuestionDefinition,
27)
28from postrfp.model.questionnaire.b36 import from_b36
29from postrfp.authorisation import perms
32@http
33def put_project_section(
34 session: Session,
35 user: User,
36 project_id: int,
37 section_id: int,
38 edit_sec_doc: serial.EditableSection,
39):
40 """Update title or description for the given Section"""
41 section = fetch.section_by_id(session, section_id)
42 project = section.project
43 authorise.check(
44 user, perms.PROJECT_EDIT_COSMETIC, project=project, section_id=section.id
45 )
47 evt = AuditEvent.create(
48 session,
49 evt_types.SECTION_UPDATED,
50 project=project,
51 user_id=user.id,
52 org_id=user.organisation.id,
53 object_id=section.id,
54 private=True,
55 )
57 for attr_name in ("description", "title"):
58 new_val = getattr(edit_sec_doc, attr_name, None)
59 if new_val is not None:
60 old_val = getattr(section, attr_name)
61 if new_val != old_val:
62 evt.add_change(attr_name, old_val, new_val)
63 setattr(section, attr_name, new_val)
65 session.add(evt)
66 return serial.Section.model_validate(section)
69@http
70def post_project_section(
71 session: Session,
72 user: User,
73 project_id: int,
74 section_id: int,
75 edit_sec_doc: serial.EditableSection,
76) -> serial.Section:
77 """Create a new Section"""
78 parent = fetch.section_by_id(session, section_id)
79 project = fetch.project(session, project_id)
80 authorise.check(
81 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section_id
82 )
84 new_section = Section(
85 title=edit_sec_doc.title, description=edit_sec_doc.description
86 )
88 parent.subsections.append(new_section)
89 parent.renumber()
91 evt = AuditEvent.create(
92 session,
93 evt_types.SECTION_CREATED,
94 project=project,
95 user_id=user.id,
96 org_id=user.organisation.id,
97 object_id=new_section.id,
98 private=True,
99 )
100 evt.add_change("title", None, new_section.title)
101 if edit_sec_doc.description:
102 evt.add_change("description", None, new_section.description)
103 session.add(evt)
104 return serial.Section.model_validate(new_section)
107@http
108def delete_project_section(
109 session: Session, user: User, project_id: int, section_id: int
110):
111 """
112 Delete the given Section and all questions and subsections contained within that
113 section.
115 @permission PROJECT_SAVE_QUESTIONNAIRE
116 """
117 section = fetch.section_by_id(session, section_id)
118 project = section.project
120 authorise.check(
121 user, perms.PROJECT_SAVE_QUESTIONNAIRE, project=project, section_id=section_id
122 )
124 return update.delete_project_section(session, user, project, section)
127@http
128def put_project_movesection(
129 session: Session, user: User, project_id: int, move_section_doc: serial.MoveSection
130) -> serial.Section:
131 """
132 Move the section given by section_id to a location with the same project given by
133 new_parent_id - the ID of the new Parent section
134 """
135 project = fetch.project(session, project_id)
136 new_parent_id = move_section_doc.new_parent_id
137 section_id = move_section_doc.section_id
139 authorise.check(
140 user,
141 perms.PROJECT_SAVE_QUESTIONNAIRE,
142 project=project,
143 section_id=section_id,
144 deny_restricted=True,
145 )
146 authorise.check(
147 user,
148 perms.PROJECT_SAVE_QUESTIONNAIRE,
149 project=project,
150 section_id=new_parent_id,
151 deny_restricted=True,
152 )
154 section = project.sections.filter_by(id=section_id).one()
155 parent_section = project.sections.filter_by(id=new_parent_id).one()
157 if section is parent_section:
158 m = "section_id and new_parent_id must be different values"
159 raise QuestionnaireStructureException(m)
161 if section.parent_id == new_parent_id:
162 raise ValueError(
163 "parent_section_id provided cannot be the same as the current parent_id"
164 )
166 section.parent_id = new_parent_id
167 session.flush()
168 if parent_section.parent:
169 parent_section.parent.renumber()
170 else:
171 parent_section.renumber()
172 session.refresh(section)
173 return serial.Section.model_validate(section)
176@http
177def put_project_section_children(
178 session: Session,
179 user: User,
180 project_id: int,
181 section_id: int,
182 child_nodes_doc: serial.SectionChildNodes,
183) -> serial.SummarySection:
184 """
185 Set the contents of the Section by providing a list of section IDs *or* question IDs.
187 This method can be used to move *existing* questions or sections to a new parent section or
188 to re-order existing section contents. It cannot be used to move questions or sections from a
189 different project.
191 Child objects (Section or Question) which exist within the current section but whose ID is not
192 included in the JSON body parameter are termed 'orphans'. If 'delete_orphans' is true, then
193 such objects will be deleted, i.e. the contents of the current section will be exactly the same
194 as that provided by the question_ids or section_ids parameter.
196 The default value for 'delete_orphans' is false - this value should only be overridden with
197 care.
199 N.B. The JSON body object should contain either question_ids or section_ids arrays, but
200 not both.
201 """
202 from postrfp.shared.movenodes import (
203 SectionsMover,
204 QuestionsMover,
205 AbstractNodeMover,
206 )
208 project = fetch.project(session, project_id)
209 parent: Section = project.sections.filter_by(id=section_id).one()
210 authorise.check(user, perms.PROJECT_EDIT, project=project, section_id=parent.id)
212 section_ids = child_nodes_doc.section_ids
213 question_ids = child_nodes_doc.question_ids
215 mover: AbstractNodeMover
217 if section_ids:
218 if parent.id in set(section_ids):
219 raise ValueError(
220 f"Parent section ID {parent.id} (from URL param) cannot be assigned under section_ids"
221 )
222 mover = SectionsMover(
223 session=session,
224 user=user,
225 project=project,
226 parent=parent,
227 provided_ids=section_ids,
228 delete_orphans=child_nodes_doc.delete_orphans,
229 )
230 elif question_ids:
231 mover = QuestionsMover(
232 session=session,
233 user=user,
234 project=project,
235 parent=parent,
236 provided_ids=question_ids,
237 delete_orphans=child_nodes_doc.delete_orphans,
238 )
239 else:
240 raise ValueError("Either section_ids or question_ids must be provided")
242 mover.execute()
244 return serial.SummarySection.model_validate(parent)
247@http
248def get_project_section(
249 session: Session, user: User, project_id: int, section_id: int
250) -> serial.SummarySection:
251 """
252 Get the Section with the given ID together with questions and subsections contained
253 within the Section.
254 """
255 project = fetch.project(session, project_id)
256 section = fetch.section_of_project(project, section_id)
257 authorise.check(
258 user,
259 perms.PROJECT_VIEW_QUESTIONNAIRE,
260 project=project,
261 section_id=section.id,
262 deny_restricted=False,
263 )
265 sec_doc = serial.SummarySection(
266 id=section.id,
267 parent_id=section.parent_id,
268 number=section.number,
269 title=section.title,
270 description=section.description,
271 )
273 sq = section.questions_query.join(
274 QuestionDefinition,
275 QuestionInstance.question_def_id == QuestionDefinition.id,
276 ).with_entities(
277 QuestionInstance.id, QuestionDefinition.title, QuestionInstance.b36_number
278 )
279 sec_doc.questions = [
280 serial.Node(id=r.id, title=r.title, number=from_b36(r.b36_number)) for r in sq
281 ]
283 for subsec in fetch.visible_subsections_query(section, user):
284 sec_doc.subsections.append(
285 serial.Node(id=subsec.id, title=subsec.title, number=subsec.number)
286 )
288 return sec_doc
291@http_etag
292def get_project_treenodes(
293 session: Session, user: User, project_id: int
294) -> List[serial.ProjectNode]:
295 """
296 N.B. - this method is deprecated. Use get_project_nodes instead.
298 Get an array of all sections, subsections and questions for the given project id.
300 Each section or question is a "node" in the questionnaire tree structure. The array
301 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
302 """
303 project = fetch.project(session, project_id)
304 authorise.check(
305 user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project, deny_restricted=True
306 )
307 nodes = []
309 for node in fetch.light_nodes(session, project_id):
310 nodes.append(
311 serial.ProjectNode(
312 id=node.id,
313 title=node.title,
314 type=node.type,
315 parent_id=node.parent_id,
316 number=from_b36(node.b36_number),
317 position=node.position,
318 depth=int(node.depth),
319 )
320 )
321 return nodes
324@http
325def get_project_nodes(
326 session: Session,
327 user: User,
328 project_id: int,
329 q_section_id: int,
330 with_ancestors: bool,
331) -> List[serial.ProjectNode]:
332 """
333 Get an array of subsections and questions for the given project id and section id.
334 If section ID is not provided then results for the root section of the project are returned.
336 If ancestors parameter is true or 1, then nodes of all direct ancestors and siblings of those
337 ancestors are returned, facilitating the respresentation of a tree structure.
339 Each section or question is a "node" in the questionnaire. The array
340 is sorted in document order: 1.1.1, 1.1.2, 1.2.1, 1.2.2 etc.
341 """
342 project = fetch.project(session, project_id)
343 authorise.check(
344 user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project, deny_restricted=True
345 )
347 if q_section_id is None:
348 q_section_id = project.section_id
349 sec = (
350 session.query(Section)
351 .filter(Section.id == q_section_id)
352 .filter(Section.project_id == project_id)
353 .one()
354 )
356 nodes = []
357 for node in fetch.visible_nodes(session, sec, with_ancestors=with_ancestors):
358 nodes.append(
359 serial.ProjectNode(
360 id=node.id,
361 title=node.title,
362 type=node.type,
363 parent_id=node.parent_id,
364 number=from_b36(node.b36_number),
365 position=node.position,
366 depth=int(node.depth),
367 )
368 )
369 return nodes
372@http
373def post_project_section_excel(
374 session: Session,
375 user: User,
376 project_id: int,
377 section_id: int,
378 data_upload: FieldStorage,
379) -> serial.ExcelImportResult:
380 """
381 Create questions and subsections within the given section by uploading an Excel spreadsheet
382 with a specific format of 5 columns:
384 - A: Section Title. Add the newly created question to this section, creating the Section if
385 necessary
386 - B: Question Title. Short - for reference and navigation
387 - C: Question Text. The body of the question
388 - D: Multiple Choices. Multiple choice options seperated by semi colons. A numeric value
389 in angle brackets, e.g. <5> after the option text is interpreted as the Autoscore value
390 for that option
391 - E: Comments Field header. If provided in conjunction with Multiple choice options the
392 value of this column is used to create a subheader below the multiple choice field
393 and above a text field for comments or qualifications.
396 At most 500 rows are processed. Additional rows are disregarded. If column D is empty the
397 question will consist of a single text input field.
398 """
400 from ..io import excel_import
402 project = fetch.project(session, project_id)
403 section = fetch.section_of_project(project, section_id)
404 authorise.check(
405 user,
406 perms.PROJECT_SAVE_QUESTIONNAIRE,
407 project=project,
408 section_id=section_id,
409 deny_restricted=True,
410 )
411 eqi = excel_import.ExcelQImporter(section)
412 eqi.read_questions_excel(data_upload.file)
413 if section.is_top_level:
414 section.renumber()
415 elif section.parent is not None:
416 section.parent.renumber()
418 descendant_qids = project.questions.filter(
419 QuestionInstance.b36_number.startswith(section.b36_number)
420 ).with_entities(QuestionInstance.question_def_id)
421 update_qmeta_table(session, {r[0] for r in descendant_qids})
423 return serial.ExcelImportResult(imported_count=eqi.created_count)
426@http
427def get_project_qstats(
428 session: Session, user: User, project_id: int
429) -> Annotated[dict[str, int | dict], serial.QuestionnaireStats]:
430 """
431 Retrieve statistics for counts of sections, questions and question elements (by type)
432 for the given project ID.
433 """
434 project = fetch.project(session, project_id)
435 authorise.check(user, perms.PROJECT_VIEW_QUESTIONNAIRE, project=project)
436 return fetch.questionnaire_stats(session, project_id)
439@http
440def post_project_section_import(
441 session: Session,
442 user: User,
443 project_id: int,
444 section_id: int,
445 import_section_doc: serial.SectionImportDoc,
446) -> serial.SectionImportResult:
447 """
448 Import questionnaire from previous project
449 """
451 des_project = fetch.project(session, project_id)
452 des_section = fetch.section_of_project(des_project, section_id)
454 src_project = fetch.project(session, import_section_doc.project_id)
456 authorise.check(
457 user,
458 perms.PROJECT_SAVE_QUESTIONNAIRE,
459 project=des_project,
460 section_id=des_section.id,
461 )
462 authorise.check(user, action=perms.PROJECT_ACCESS, project=src_project)
464 src_sections = []
465 for section_id in import_section_doc.section_ids:
466 src_sections.append(fetch.section_of_project(src_project, section_id))
468 src_questions = []
469 for question_id in import_section_doc.question_ids:
470 src_questions.append(fetch.question_of_project(src_project, question_id))
472 import_type = ImportType.COPY if import_section_doc.clone else ImportType.SHARE
473 if import_type == ImportType.SHARE:
474 duplicated_qis: list[QuestionInstance] = fetch.duplicated_qdefs(
475 session,
476 project_id,
477 import_section_doc.project_id,
478 src_sections,
479 src_questions,
480 )
481 if len(duplicated_qis) > 0:
482 duplicated_titles = [f"{q.number}. {q.title}" for q in duplicated_qis]
483 raise DuplicateQuestionDefinition(
484 f"Duplicate questions definition(s) {duplicated_titles}"
485 )
487 imported_sections: list[Section] = []
488 imported_questions: list[QuestionInstance] = []
489 for src_sec in src_sections:
490 sections, questions = update.import_section(
491 session, src_sec, des_section, import_type
492 )
493 imported_sections += sections
494 imported_questions += questions
495 for src_que in src_questions:
496 imported_qi: QuestionInstance = update.import_q_instance(
497 src_que, des_section, import_type
498 )
499 imported_questions.append(imported_qi)
501 session.flush()
502 if import_type == ImportType.COPY:
503 update_qmeta_table(session, {qi.question_def_id for qi in imported_questions})
505 if des_section.is_top_level:
506 des_section.renumber()
507 elif des_section.parent is not None:
508 des_section.parent.renumber()
510 return serial.SectionImportResult(
511 section_count=len(imported_sections), question_count=len(imported_questions)
512 )