Coverage for rfpy/api/endpoints/weighting.py: 99%

1""" 

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

3""" 

4 

5from rfpy.model.humans import User 

6from sqlalchemy.orm.session import Session 

7from rfpy.web.serial.models import WeightSet 

8from typing import List 

9 

10from rfpy.suxint import http 

11from rfpy.model import WeightingSet 

12from rfpy.api import fetch, validate, update 

13from rfpy.auth import perms 

14from rfpy.web import serial 

15 

16 

17@http 

18def get_project_weightings( 

19 session, user, project_id, weightset_id=None 

20) -> serial.ProjectWeightings: 

21 """ 

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

23 

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

25 section and all ancestor sections. 

26 

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

28 """ 

29 

30 project = fetch.project(session, project_id) 

31 validate.check(user, perms.PROJECT_VIEW_WEIGHTING, project=project) 

32 if weightset_id is None: 

33 name = "Default" 

34 else: 

35 name = ( 

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

37 .with_entities(WeightingSet.name) 

38 .scalar() 

39 ) 

40 

41 tw = fetch.total_weightings_dict(project, weightset_id) 

42 inst = fetch.weightings_dict(project, weightset_id) 

43 return serial.ProjectWeightings( 

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

45 total=serial.Weightings(**tw), 

46 instance=serial.Weightings(**inst), 

47 ) 

48 

49 

50@http 

51def post_project_weightings( 

52 session, user, project_id, weights_doc: serial.WeightingsDoc 

53): 

54 """ 

55 Batch saving/updating of Weightings 

56 

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

58 """ 

59 project = fetch.project(session, project_id) 

60 validate.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

61 weightset_id = weights_doc.weightset_id 

62 if weightset_id: 

63 weightset = project.weighting_sets.filter(WeightingSet.id == weightset_id).one() 

64 update.save_weightset_weightings(session, weightset, weights_doc) 

65 else: 

66 update.save_default_weightings(session, project, weights_doc) 

67 session.flush() 

68 project.delete_total_weights(weighting_set_id=weightset_id) 

69 project.save_total_weights(weighting_set_id=weightset_id) 

70 

71 

72@http 

73def get_project_weightsets(session, user, project_id) -> List[serial.WeightSet]: 

74 """ 

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

76 """ 

77 project = fetch.project(session, project_id) 

78 validate.check(user, perms.PROJECT_VIEW_WEIGHTING, project=project) 

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

80 

81 

82@http 

83def post_project_weightset( 

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

85) -> serial.WeightSet: 

86 """ 

87 Create a new Weighting Set for the given Project 

88 

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

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

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

92 """ 

93 project = fetch.project(session, project_id) 

94 validate.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

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

96 session.add(ws) 

97 session.flush() 

98 initial_value = weightset_doc.initial_value 

99 if initial_value is not None: 

100 update.set_initial_weightings(ws, initial_value) 

101 else: 

102 source_ws_id = weightset_doc.source_weightset_id 

103 source_ws = session.get(WeightingSet, source_ws_id) 

104 if source_ws is None: 

105 raise ValueError(f"Weighting set with ID {source_ws_id} not found") 

106 update.copy_weightings(source_ws, ws) 

107 

108 project.save_total_weights(weighting_set_id=ws.id) 

109 wn = ws.name or "" 

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

111 

112 

113@http 

114def put_project_weightset( 

115 session: Session, 

116 user: User, 

117 project_id: int, 

118 weightset_path_id: int, 

119 name_doc: serial.ShortName, 

120): 

121 """ 

122 Change the name of an existing Weighting Set 

123 """ 

124 project = fetch.project(session, project_id) 

125 validate.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

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

127 ws.name = name_doc.name 

128 

129 

130@http 

131def delete_project_weightset( 

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

133): 

134 """ 

135 Delete the given weighting set from the Project 

136 """ 

137 project = fetch.project(session, project_id) 

138 validate.check(user, perms.PROJECT_EDIT_WEIGHTING, project=project) 

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

140 session.delete(ws) 

141 

142 

143@http 

144def get_project_section_weightings( 

145 session, user, project_id, section_id, weightset_id 

146) -> serial.ParentedWeighting: 

147 """ 

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

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

150 subsections or questions) 

151 """ 

152 section = fetch.section_by_id(session, section_id) 

153 project = section.project 

154 

155 validate.check( 

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

157 ) 

158 

159 wdict = fetch.weightings_dict( 

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

161 ) 

162 wdict["parent_absolute_weight"] = fetch.sec_total_weighting(section, weightset_id) 

163 

164 return serial.ParentedWeighting(**wdict)