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

1""" 

2Define a network of relationships between organisations 

3""" 

4 

5from enum import Enum 

6from rfpy.model.exc import BusinessRuleViolation 

7from rfpy.model.humans import OrganisationType 

8from typing import List 

9 

10from sqlalchemy.orm import Session 

11from sqlalchemy.orm.exc import NoResultFound 

12 

13from rfpy.api import validate, fetch 

14from rfpy.auth import perms 

15from rfpy.suxint import http 

16from rfpy.model import User, Edge, RelationshipType 

17 

18from rfpy.web import serial 

19 

20 

21@http 

22def get_network(session: Session, user: User) -> List[serial.NetworkRelationship]: 

23 """ 

24 Get an array of all the Network Relationships described by your organisation 

25 """ 

26 q = fetch.edges_for_org_query(session, user.org_id) 

27 return [serial.NetworkRelationship.model_validate(r) for r in q] 

28 

29 

30@http 

31def delete_network(session: Session, user: User): 

32 """ 

33 Delete all relationships defined for this network. 

34 Relationship Types are not deleted. 

35 """ 

36 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

37 rids = {rt.id for rt in user.organisation.relationship_types} 

38 session.query(Edge).filter(Edge.relationship_id.in_(rids)).delete( 

39 synchronize_session=False 

40 ) 

41 

42 

43@http 

44def get_reltypes(session: Session, user: User) -> List[serial.RelationshipType]: 

45 """ 

46 Get an array of Relationship Types for your organisation 

47 """ 

48 return [ 

49 serial.RelationshipType(id=rt.id, name=rt.name, description=rt.description) 

50 for rt in user.organisation.relationship_types 

51 ] 

52 

53 

54@http 

55def post_reltype( 

56 session: Session, user: User, reltype_doc: serial.RelationshipType 

57) -> serial.Id: 

58 """ 

59 Create a new Relationship Type for your organisation 

60 

61 A RelationshipType must be defined for your organisation before a Relationship can 

62 be defined between two 3rd party organisations. 

63 Multiple RelationshipTypes can be defined for your organisation in order to model different 

64 types of inter-organisation relationships - e.g. 'Partner', 'Downstream Supplier', 

65 'Regulator', etc. 

66 

67 @permission MANAGE_ORGANISATION 

68 """ 

69 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

70 rt = RelationshipType(**reltype_doc.model_dump()) 

71 user.organisation.relationship_types.append(rt) 

72 session.flush() 

73 return serial.Id(id=rt.id) 

74 

75 

76@http 

77def put_reltype( 

78 session: Session, user: User, reltype_id, reltype_doc: serial.RelationshipType 

79) -> serial.Id: 

80 """ 

81 Update the name or description for the Relationship Type with the given ID 

82 """ 

83 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

84 rt = user.organisation.relationship_types.filter( 

85 RelationshipType.id == reltype_id 

86 ).one() 

87 rt.name = reltype_doc.name 

88 rt.description = reltype_doc.description 

89 

90 return serial.Id(id=rt.id) 

91 

92 

93@http 

94def delete_reltype(session: Session, user: User, reltype_id: int) -> serial.Id: 

95 """ 

96 Delete the Relationship Type with the given ID together with all 

97 relationships of that type. Underlying Organisations are not deleted. 

98 """ 

99 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

100 rt = user.organisation.relationship_types.filter( 

101 RelationshipType.id == reltype_id 

102 ).one() 

103 session.delete(rt) 

104 return serial.Id(id=rt.id) 

105 

106 

107@http 

108def post_relationship( 

109 session: Session, user: User, relationship_doc: serial.Relationship 

110) -> None: 

111 """ 

112 Create a new Relationship between two organisations 

113 """ 

114 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

115 reltype_id = relationship_doc.reltype_id 

116 rt = user.organisation.relationship_types.filter( 

117 RelationshipType.id == reltype_id 

118 ).one() 

119 from_org = fetch.organisation(session, relationship_doc.from_org_id) 

120 to_org = fetch.organisation(session, relationship_doc.to_org_id) 

121 edge = Edge(to_org=to_org, from_org=from_org, relationship_type=rt) 

122 session.add(edge) 

123 

124 

125@http 

126def delete_relationship( 

127 session: Session, user: User, relationship_doc: serial.Relationship 

128) -> None: 

129 """ 

130 Delete a new Relationship between two organisations 

131 """ 

132 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

133 reltype_id = relationship_doc.reltype_id 

134 rt = user.organisation.relationship_types.filter( 

135 RelationshipType.id == reltype_id 

136 ).one() 

137 from_org = fetch.organisation(session, relationship_doc.from_org_id) 

138 to_org = fetch.organisation(session, relationship_doc.to_org_id) 

139 edge = ( 

140 session.query(Edge) 

141 .filter_by(to_org_id=to_org.id, from_org_id=from_org.id, relationship_id=rt.id) 

142 .one() 

143 ) 

144 session.delete(edge) 

145 

146 

147class StandardRelTypes(str, Enum): 

148 CONSULTING_ENGAGEMENT = "Consults For" 

149 CONSULTING_PARTNERSHIP = "Partners With" 

150 SUPPLIES = "Supplies" 

151 VENDOR_EVALUATION = "Evaluates" 

152 

153 

154@http 

155def post_network_project( 

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

157) -> List[serial.NetworkRelationship]: 

158 """ 

159 Generate a network of relationships between vendors and participants for the project ID 

160 provided. Creates default Relationship Types for standard RFP project relationships. 

161 """ 

162 

163 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

164 if user.organisation.type != OrganisationType.CONSULTANT: 

165 raise BusinessRuleViolation( 

166 "Action only permitted for Consultant Organisations" 

167 ) 

168 project = fetch.project(session, project_id=project_id) 

169 validate.check(user, action=perms.PROJECT_ACCESS, project=project) 

170 org = user.organisation 

171 

172 rt_lookup = dict() 

173 for SRT in StandardRelTypes: 

174 try: 

175 rel = org.relationship_types.filter_by(name=SRT.value).one() 

176 except NoResultFound: 

177 rel = RelationshipType(org_id=org.id, name=SRT.value) 

178 session.add(rel) 

179 rt_lookup[SRT] = rel 

180 

181 # Populate Edge objects in the Session to facilitate merge() operations later 

182 fetch.edges_for_org_query(session, user.org_id).all() 

183 

184 for participant in project.participants: 

185 if participant.organisation is org: 

186 continue 

187 if participant.organisation.is_consultant: 

188 rel = rt_lookup[StandardRelTypes.CONSULTING_PARTNERSHIP] 

189 else: 

190 rel = rt_lookup[StandardRelTypes.CONSULTING_ENGAGEMENT] 

191 edge = Edge( 

192 from_org_id=org.id, to_org_id=participant.org_id, relationship_id=rel.id 

193 ) 

194 session.merge(edge) 

195 

196 respondent_id_set = {i.respondent_id for i in project.issues} 

197 

198 for respondent_id in respondent_id_set: 

199 if respondent_id is None or respondent_id == org.id: 

200 continue 

201 

202 buyer_id = project.org_id # Owner org assumed to be the buyer 

203 

204 rel = rt_lookup[StandardRelTypes.SUPPLIES] 

205 session.merge( 

206 Edge(from_org_id=respondent_id, to_org_id=buyer_id, relationship_id=rel.id) 

207 ) 

208 

209 evaluates_rel = rt_lookup[StandardRelTypes.VENDOR_EVALUATION] 

210 session.merge( 

211 Edge( 

212 from_org_id=org.id, 

213 to_org_id=respondent_id, 

214 relationship_id=evaluates_rel.id, 

215 ) 

216 ) 

217 

218 return get_network(session, user)