Coverage for postrfp/shared/fetch/weightq.py: 100%

1import logging 

2from typing import ( 

3 Optional, 

4 List, 

5) 

6from typing_extensions import TypedDict 

7 

8from postrfp.model import ( 

9 Project, 

10 Section, 

11 QuestionInstance, 

12) 

13from postrfp.model.questionnaire.weightings import ( 

14 TotalWeighting, 

15 Weighting, 

16) 

17 

18log = logging.getLogger(__name__) 

19 

20 

21# TypedDict definitions for type safety 

22class QuestionWeight(TypedDict): 

23 question_instance_id: int 

24 weight: float 

25 

26 

27class SectionWeight(TypedDict): 

28 section_id: int 

29 weight: float 

30 

31 

32class WeightingsData(TypedDict): 

33 questions: List[QuestionWeight] 

34 sections: List[SectionWeight] 

35 

36 

37def _ws_weights( 

38 project: Project, weightset_id: int, parent_section_id: Optional[int] = None 

39) -> WeightingsData: 

40 """Get Weights for Questions and Sections for the given weightset_id""" 

41 

42 questions: List[QuestionWeight] = [] 

43 sections: List[SectionWeight] = [] 

44 

45 # Get all sections in scope (with explicit weights via LEFT JOIN) 

46 section_query = project.sections 

47 if parent_section_id is not None: 

48 section_query = section_query.filter(Section.parent_id == parent_section_id) 

49 

50 # LEFT JOIN to get explicit weights, defaulting to 1.0 for missing records 

51 # NOTE: Explicit ordering is important for deterministic API responses. 

52 # Without an ORDER BY, different DB backends / query plans can yield a 

53 # non-deterministic sequence of rows (especially when outer joins and 

54 # missing rows are involved). This was causing fragile snapshot tests 

55 # where e.g. section_id 1 occasionally appeared at the end of the list. 

56 section_weights = ( 

57 section_query.outerjoin( 

58 Weighting, 

59 (Weighting.section_id == Section.id) 

60 & (Weighting.weighting_set_id == weightset_id), 

61 ) 

62 .with_entities(Section.id, Weighting.value) 

63 .order_by(Section.id) 

64 ) 

65 

66 for section_id, weight_value in section_weights: 

67 weight = float(weight_value) if weight_value is not None else 1.0 

68 sections.append({"section_id": section_id, "weight": weight}) 

69 

70 # Get all questions in scope (with explicit weights via LEFT JOIN) 

71 question_query = project.questions 

72 if parent_section_id is not None: 

73 question_query = question_query.filter( 

74 QuestionInstance.section_id == parent_section_id 

75 ) 

76 

77 # LEFT JOIN to get explicit weights, defaulting to 1.0 for missing records 

78 question_weights = ( 

79 question_query.outerjoin( 

80 Weighting, 

81 (Weighting.question_instance_id == QuestionInstance.id) 

82 & (Weighting.weighting_set_id == weightset_id), 

83 ) 

84 .with_entities(QuestionInstance.id, Weighting.value) 

85 .order_by(QuestionInstance.id) 

86 ) 

87 

88 question_weights = question_weights.order_by(QuestionInstance.id) 

89 

90 for question_id, weight_value in question_weights: 

91 weight = float(weight_value) if weight_value is not None else 1.0 

92 questions.append({"question_instance_id": question_id, "weight": weight}) 

93 

94 return {"questions": questions, "sections": sections} 

95 

96 

97def _default_weights( 

98 project: Project, parent_section_id: Optional[int] = None 

99) -> WeightingsData: 

100 """Default weights for all sections and questions in the project""" 

101 return _ws_weights( 

102 project, 

103 project.get_or_create_default_weighting_set_id(), 

104 parent_section_id=parent_section_id, 

105 ) 

106 

107 

108def weightings_dict( 

109 project: Project, 

110 weightset_id: Optional[int] = None, 

111 parent_section_id: Optional[int] = None, 

112) -> WeightingsData: 

113 """ 

114 Get a dictionary of weightings for all sections and 

115 questions in the given project, 

116 { 

117 questions: [{question_instance_id: question_weight}] 

118 sections: [{section_id: section_weight}] 

119 }, 

120 for the given project and weightset_id 

121 """ 

122 if weightset_id is None: 

123 return _default_weights(project, parent_section_id=parent_section_id) 

124 else: 

125 return _ws_weights(project, weightset_id, parent_section_id=parent_section_id) 

126 

127 

128def total_weightings_dict(project: Project, weightset_id: int) -> WeightingsData: 

129 """ 

130 Get a dictionary of *total* weightings for each section 

131 and question in the given project { 

132 weightset: {id, name} 

133 questions: {question_instance_id: total_question_weight} 

134 sections: {section_id: total_section_weight} 

135 }, 

136 for the given project and weightset_id 

137 """ 

138 q = ( 

139 project.total_weightings.filter(TotalWeighting.weighting_set_id == weightset_id) 

140 .with_entities( 

141 TotalWeighting.question_instance_id, 

142 TotalWeighting.section_id, 

143 # Used by scoring and reporting systems throughout the application 

144 TotalWeighting.weight, 

145 ) 

146 .order_by(TotalWeighting.question_instance_id, TotalWeighting.section_id) 

147 ) 

148 

149 res: WeightingsData = {"questions": [], "sections": []} 

150 sections: List[SectionWeight] = res["sections"] 

151 questions: List[QuestionWeight] = res["questions"] 

152 

153 for tw in q: 

154 weight_value = tw.weight 

155 

156 if tw.section_id: # zero is the default 

157 sections.append({"section_id": tw.section_id, "weight": weight_value}) 

158 else: 

159 questions.append( 

160 { 

161 "question_instance_id": tw.question_instance_id, 

162 "weight": weight_value, 

163 } 

164 ) 

165 

166 return res