Coverage for postrfp / model / questionnaire / renumber.py: 100%

1""" 

2Section and Question Numbering Module 

3 

4This module provides efficient functions for renumbering sections and questions 

5in hierarchical trees. It uses a base36-encoded materialized path approach where 

6each two characters in a node's b36_number represent a level in the hierarchy. 

7 

8The algorithm works as follows: 

91. Fetch all relevant sections and questions as lightweight tuples 

102. Build a parent-to-children lookup dictionary to represent the hierarchy 

113. Traverse the hierarchy to generate new node numbers based on position 

124. Apply all updates to the database in a single bulk operation 

13 

14Features: 

15- Supports renumbering from any section in the hierarchy (not just the root) 

16- Uses efficient queries with materialized path pattern matching 

17- Performs bulk updates for better performance with large trees 

18 

19Example: 

20 To renumber a section and all its descendants: 

21 

22 ```python 

23 from postrfp.model.questionnaire.renumber import renumber_tree 

24 from sqlalchemy.orm import Session 

25 

26 session = Session() 

27 section = session.get(Section, section_id) 

28 renumber_tree(session, section) 

29 ``` 

30""" 

31 

32from sqlalchemy.orm import Session 

33from sqlalchemy import update, select, func, Integer 

34 

35from postrfp.model import Section, QuestionInstance 

36from postrfp.model.questionnaire.b36 import materialised_paths 

37 

38 

39def fetch_section_tuples(session: Session, section): 

40 """ 

41 Fetch the section and all its descendants as lightweight tuples, 

42 using recursive query based on parent_id relationship instead of b36_number. 

43 

44 Args: 

45 session: SQLAlchemy session 

46 section: The section to start renumbering from 

47 """ 

48 

49 # Define the base table for sections 

50 sections = Section.__table__ 

51 

52 # Base case: the section itself 

53 base_query = ( 

54 select( 

55 sections.c.id.cast(Integer).label("id"), 

56 sections.c.parent_id.cast(Integer).label("parent_id"), 

57 sections.c.position.cast(Integer).label("position"), 

58 ) 

59 .where(sections.c.id == section.id) 

60 .where(sections.c.project_id == section.project_id) 

61 ) 

62 

63 # Create the recursive CTE 

64 descendants_cte = base_query.cte(name="descendants", recursive=True) 

65 

66 # Recursive case: all children of sections already in the result 

67 recursive_query = ( 

68 select( 

69 sections.c.id.cast(Integer).label("id"), 

70 sections.c.parent_id.cast(Integer).label("parent_id"), 

71 sections.c.position.cast(Integer).label("position"), 

72 ) 

73 .select_from( 

74 sections.join(descendants_cte, sections.c.parent_id == descendants_cte.c.id) 

75 ) 

76 .where(sections.c.project_id == section.project_id) 

77 ) 

78 

79 # Union the base case and recursive case 

80 descendants_cte = descendants_cte.union_all(recursive_query) 

81 

82 # Final query to select from the CTE 

83 final_query = select( 

84 descendants_cte.c.id, descendants_cte.c.parent_id, descendants_cte.c.position 

85 ).order_by(descendants_cte.c.parent_id, descendants_cte.c.position) 

86 

87 result = session.execute(final_query) 

88 return result.all() 

89 

90 

91def save_section_numbers(session, section_numbers): 

92 """Apply the generated node numbers to the database using ORM updates to ensure proper type conversion""" 

93 

94 # Get all the sections that need updating 

95 sections_to_update = ( 

96 session.query(Section).filter(Section.id.in_(section_numbers.keys())).all() 

97 ) 

98 

99 # Update each section through the ORM to ensure type conversion 

100 for section in sections_to_update: 

101 section.b36_number = section_numbers[section.id] 

102 

103 session.flush() 

104 

105 # Verify the values were saved correctly 

106 for section in sections_to_update: 

107 session.refresh(section) 

108 

109 

110def update_qinstance_numbers(session: Session, section_ids): 

111 """ 

112 Derive "b36_number" values for all QuestionInstances based on the 

113 b36_number of parent section b36_number 

114 """ 

115 # Create a subquery to get the section numbers 

116 section_query = ( 

117 select(Section.id, Section.b36_number) 

118 .where(Section.id.in_(section_ids)) 

119 .subquery("section_numbers") 

120 ) 

121 stmt = ( 

122 update(QuestionInstance) 

123 .where(QuestionInstance.section_id.in_(section_ids)) 

124 .values( 

125 b36_number=func.concat( 

126 section_query.c.b36_number, 

127 func.lpad(func.conv(QuestionInstance.position, 10, 36), 2, "0"), 

128 ) 

129 ) 

130 .where(QuestionInstance.section_id == section_query.c.id) 

131 ) 

132 session.execute(stmt) 

133 

134 

135def renumber_tree(session: Session, root_section: Section) -> None: 

136 """ 

137 Efficiently renumber a section tree and all its questions. 

138 Works with any section at any level of the hierarchy. 

139 

140 Args: 

141 session: The SQLAlchemy session 

142 section: The Section object to renumber from (can be at any level) 

143 """ 

144 # Fetch all sections 

145 sec_nodes = fetch_section_tuples(session, root_section) 

146 # build a dict of {id: b36_number} (path) - this is the main algorithm 

147 sec_paths = materialised_paths(sec_nodes, root_section) 

148 # save the new b36_number values to Sections by id 

149 save_section_numbers(session, sec_paths) 

150 # Use the section b36_numbers to derive the question b36_numbers (in sql) 

151 update_qinstance_numbers(session, sec_paths.keys())