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

1''' 

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

3''' 

4from rfpy.model.humans import User 

5from sqlalchemy.orm.session import Session 

6from rfpy.web.serial.models import WeightSet 

7from typing import List 

8 

9from rfpy.suxint import http 

10from rfpy.model import ( 

11 WeightingSet 

12) 

13from rfpy.api import fetch, validate, update 

14from rfpy.auth import perms 

15from rfpy.web import serial 

16 

17 

18@http 

19def get_project_weightings(session, user, 

20 project_id, 

21 weightset_id=None) -> serial.ProjectWeightings: 

22 ''' 

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

24 

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

26 section and all ancestor sections. 

27 

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

29 ''' 

30 

31 project = fetch.project(session, project_id) 

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

33 if weightset_id is None: 

34 name = 'Default' 

35 else: 

36 name = project.weighting_sets\ 

37 .filter(WeightingSet.id == weightset_id)\ 

38 .with_entities(WeightingSet.name)\ 

39 .scalar() 

40 res = { 

41 'weightset': {'name': name, 'id': weightset_id}, 

42 'total': fetch.total_weightings_dict(project, weightset_id), 

43 'instance': fetch.weightings_dict(project, weightset_id) 

44 } 

45 return res 

46 

47 

48@http 

49def post_project_weightings(session, user, project_id, weights_doc: serial.WeightingsDoc): 

50 ''' 

51 Batch saving/updating of Weightings 

52 

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

54 ''' 

55 project = fetch.project(session, project_id) 

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

57 weightset_id = weights_doc["weightset_id"] 

58 if weightset_id: 

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

60 update.save_weightset_weightings(session, weightset, weights_doc) 

61 else: 

62 update.save_default_weightings(session, project, weights_doc) 

63 session.flush() 

64 project.delete_total_weights(weighting_set_id=weightset_id) 

65 project.save_total_weights(weighting_set_id=weightset_id) 

66 

67 

68@http 

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

70 ''' 

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

72 ''' 

73 project = fetch.project(session, project_id) 

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

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

76 

77 

78@http 

79def post_project_weightset( 

80 session: Session, 

81 user: User, 

82 project_id: int, 

83 weightset_doc: serial.NewWeightSet) -> serial.WeightSet: 

84 ''' 

85 Create a new Weighting Set for the given Project 

86 

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

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

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

90 ''' 

91 project = fetch.project(session, project_id) 

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

93 ws = WeightingSet(project=project, name=weightset_doc['name']) 

94 session.add(ws) 

95 session.flush() 

96 initial_value = weightset_doc.get('initial_value', None) 

97 if initial_value is not None: 

98 update.set_initial_weightings(ws, initial_value) 

99 else: 

100 source_ws_id = weightset_doc['source_weightset_id'] 

101 source_ws = session.query(WeightingSet).get(source_ws_id) 

102 if source_ws is None: 

103 raise ValueError(f'Weighting set with ID {source_ws_id} not found') 

104 update.copy_weightings(source_ws, ws) 

105 

106 project.save_total_weights(weighting_set_id=ws.id) 

107 return WeightSet(name=ws.name, id=ws.id) 

108 

109 

110@http 

111def put_project_weightset( 

112 session: Session, 

113 user: User, 

114 project_id: int, 

115 weightset_path_id: int, 

116 name_doc: serial.ShortName): 

117 ''' 

118 Change the name of an existing Weighting Set 

119 ''' 

120 project = fetch.project(session, project_id) 

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

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

123 ws.name = name_doc['name'] 

124 

125 

126@http 

127def delete_project_weightset( 

128 session: Session, 

129 user: User, 

130 project_id: int, 

131 weightset_path_id: int): 

132 ''' 

133 Delete the given weighting set from the Project 

134 ''' 

135 project = fetch.project(session, project_id) 

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

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

138 session.delete(ws) 

139 

140 

141@http 

142def get_project_section_weightings(session, 

143 user, 

144 project_id, 

145 section_id, 

146 weightset_id) -> 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(user, perms.PROJECT_VIEW_WEIGHTING, project=project, section_id=section.id) 

156 

157 wdict = fetch.weightings_dict(project, weightset_id=weightset_id, parent_section_id=section.id) 

158 wdict['parent_absolute_weight'] = fetch.sec_total_weighting(section, weightset_id) 

159 return wdict