Coverage for postrfp/buyer/api/endpoints/weighting.py: 97%

1""" 

2Assign weightings to Sections and Questions. Manage named weighting sets. 

3""" 

4 

5from typing import cast 

6from decimal import Decimal 

7 

8from sqlalchemy.orm.session import Session 

9from sqlalchemy.orm.exc import NoResultFound 

10 

11from postrfp.shared import fetch, update 

12from postrfp.shared.serial.models import WeightSet 

13from postrfp.model.humans import User 

14from postrfp.shared.decorators import http 

15from postrfp.model.questionnaire.weightings import WeightingSet 

16from postrfp.buyer.api import authorise 

17from postrfp.authorisation import perms 

18from postrfp.shared import serial 

19 

20 

21@http 

22def get_project_weightings( 

23 session: Session, user: User, project_id: int, weightset_id: int | None = None 

24) -> serial.ProjectWeightings: 

25 """ 

26 Get total and instance(local) weightings for questions and sections in the current project. 

27 

28 Total weightings are the normalised, hierarchical weights reflecting the weight of the parent 

29 section and all ancestor sections. 

30 

31 Instance weighings are the weightings that the user assigns to a question or section. 

32 """ 

33 

34 project = fetch.project(session, project_id) 

35 authorise.check(user, perms.PROJECT_VIEW_WEIGHTING, project=project) 

36 

37 # Resolve None weightset_id to the actual default weighting set ID 

38 if weightset_id is None: 

39 weightset_id = project.get_or_create_default_weighting_set_id() 

40 name = "Default" 

41 else: 

42 name = ( 

43 project.weighting_sets.filter(WeightingSet.id == weightset_id) 

44 .with_entities(WeightingSet.name) 

45 .scalar() 

46 ) 

47 

48 tw = fetch.total_weightings_dict(project, weightset_id) 

49 inst = fetch.weightings_dict(project, weightset_id) 

50 

51 # Use cast to ensure the types match the expected serial models 

52 # this only affects type checking, not runtime behavior 

53 return serial.ProjectWeightings( 

54 weightset=serial.WeightSet(name=name, id=weightset_id), 

55 total=serial.Weightings(**cast(dict, tw)), 

56 instance=serial.Weightings(**cast(dict, inst)), 

57 ) 

58 

59 

60@http 

61def post_project_weightings( 

62 session: Session, user: User, project_id: int, weights_doc: serial.WeightingsDoc 

63): 

64 """ 

65 Batch saving/updating of Weightings 

66 

67 To save default weightings set the value of weightset_id to null. 

68 """ 

69 project = fetch.project(session, project_id) 

70 authorise.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

71 weightset_id = weights_doc.weightset_id 

72 if weightset_id: 

73 try: 

74 weightset = project.weighting_sets.filter( 

75 WeightingSet.id == weightset_id 

76 ).one() 

77 except NoResultFound: 

78 raise ValueError( 

79 f"Weighting Set with ID {weightset_id} does not exist in project {project_id}." 

80 ) 

81 update.save_weightset_weightings(session, weightset, weights_doc) 

82 else: 

83 update.save_default_weightings(session, project, weights_doc) 

84 session.flush() 

85 project.delete_total_weights(weighting_set_id=weightset_id) 

86 # Use the new CTE-based calculation by default 

87 project.save_total_weights(weighting_set_id=weightset_id) 

88 

89 

90@http 

91def get_project_weightsets( 

92 session: Session, user: User, project_id: int 

93) -> list[serial.WeightSet]: 

94 """ 

95 Get an array of Weighting Set objects for the given Project 

96 """ 

97 project = fetch.project(session, project_id) 

98 authorise.check(user, perms.PROJECT_VIEW_WEIGHTING, project=project) 

99 return [ws.as_dict() for ws in project.weighting_sets] 

100 

101 

102@http 

103def post_project_weightset( 

104 session: Session, user: User, project_id: int, weightset_doc: serial.NewWeightSet 

105) -> serial.WeightSet: 

106 """ 

107 Create a new Weighting Set for the given Project 

108 

109 Weightings will be created for each question and section in the project. The value of 

110 these weightings will be set to the initial_value field in the json body, if given, 

111 or will be copied from the weightset whose ID is given by source_weightset_id. 

112 """ 

113 project = fetch.project(session, project_id) 

114 authorise.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

115 ws = WeightingSet(project=project, name=weightset_doc.name) 

116 session.add(ws) 

117 session.flush() 

118 initial_value = weightset_doc.initial_value 

119 if initial_value is not None: 

120 update.set_initial_weightings(ws, Decimal(initial_value)) 

121 else: 

122 source_ws_id = weightset_doc.source_weightset_id 

123 source_ws = session.get_one(WeightingSet, source_ws_id) 

124 update.copy_weightings(source_ws, ws) 

125 

126 project.save_total_weights(weighting_set_id=ws.id) 

127 wn = ws.name or "" 

128 return WeightSet(name=wn, id=ws.id) 

129 

130 

131@http 

132def put_project_weightset( 

133 session: Session, 

134 user: User, 

135 project_id: int, 

136 weightset_path_id: int, 

137 name_doc: serial.ShortName, 

138): 

139 """ 

140 Change the name of an existing Weighting Set 

141 """ 

142 project = fetch.project(session, project_id) 

143 authorise.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

144 ws = project.weighting_sets.filter(WeightingSet.id == weightset_path_id).one() 

145 ws.name = name_doc.name 

146 

147 

148@http 

149def delete_project_weightset( 

150 session: Session, user: User, project_id: int, weightset_path_id: int 

151): 

152 """ 

153 Delete the given weighting set from the Project 

154 """ 

155 project = fetch.project(session, project_id) 

156 authorise.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

157 ws = project.weighting_sets.filter(WeightingSet.id == weightset_path_id).one() 

158 session.delete(ws) 

159 

160 

161@http 

162def get_project_section_weightings( 

163 session: Session, user: User, project_id: int, section_id: int, weightset_id: int 

164) -> serial.ParentedWeighting: 

165 """ 

166 Get weightings for the Sections or Questions in the given section, together with the 

167 absolute weighting of parent section (allowing absolute weights to be calculated for 

168 subsections or questions) 

169 """ 

170 section = fetch.section_by_id(session, section_id) 

171 project = section.project 

172 

173 authorise.check( 

174 user, perms.PROJECT_VIEW_WEIGHTING, project=project, section_id=section.id 

175 ) 

176 

177 wdict = fetch.weightings_dict( 

178 project, weightset_id=weightset_id, parent_section_id=section.id 

179 ) 

180 

181 # Need to create a new dict to add parent_absolute_weight 

182 result_dict = { 

183 "questions": wdict["questions"], 

184 "sections": wdict["sections"], 

185 "parent_absolute_weight": fetch.sec_total_weighting(section, weightset_id), 

186 } 

187 

188 return serial.ParentedWeighting(**cast(dict, result_dict))