Coverage for postrfp/buyer/api/endpoints/network.py: 100%

1""" 

2Define a network of relationships between organisations 

3""" 

4 

5from enum import Enum 

6from typing import List 

7 

8from sqlalchemy.orm import Session 

9from sqlalchemy.orm.exc import NoResultFound 

10 

11from postrfp.buyer.api import authorise 

12from postrfp.authorisation import perms 

13from postrfp.shared import fetch, serial 

14from postrfp.shared.decorators import http 

15from postrfp.model import User, Edge, RelationshipType 

16from postrfp.model.exc import BusinessRuleViolation 

17from postrfp.model.humans import OrganisationType 

18 

19 

20@http 

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

22 """ 

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

24 """ 

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

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

27 

28 

29@http 

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

31 """ 

32 Delete all relationships defined for this network. 

33 Relationship Types are not deleted. 

34 """ 

35 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

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

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

38 synchronize_session=False 

39 ) 

40 

41 

42@http 

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

44 """ 

45 Get an array of Relationship Types for your organisation 

46 """ 

47 return [ 

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

49 for rt in user.organisation.relationship_types 

50 ] 

51 

52 

53@http 

54def post_reltype( 

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

56) -> serial.Id: 

57 """ 

58 Create a new Relationship Type for your organisation 

59 

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

61 be defined between two 3rd party organisations. 

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

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

64 'Regulator', etc. 

65 

66 @permission MANAGE_ORGANISATION 

67 """ 

68 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

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

70 user.organisation.relationship_types.append(rt) 

71 session.flush() 

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

73 

74 

75@http 

76def put_reltype( 

77 session: Session, user: User, reltype_id: int, reltype_doc: serial.RelationshipType 

78) -> serial.Id: 

79 """ 

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

81 """ 

82 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

83 rt = user.organisation.relationship_types.filter( 

84 RelationshipType.id == reltype_id 

85 ).one() 

86 rt.name = reltype_doc.name 

87 rt.description = reltype_doc.description 

88 

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

90 

91 

92@http 

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

94 """ 

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

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

97 """ 

98 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

99 rt = user.organisation.relationship_types.filter( 

100 RelationshipType.id == reltype_id 

101 ).one() 

102 session.delete(rt) 

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

104 

105 

106@http 

107def post_relationship( 

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

109) -> None: 

110 """ 

111 Create a new Relationship between two organisations 

112 """ 

113 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

114 reltype_id = relationship_doc.reltype_id 

115 rt = user.organisation.relationship_types.filter( 

116 RelationshipType.id == reltype_id 

117 ).one() 

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

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

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

121 session.add(edge) 

122 

123 

124@http 

125def delete_relationship( 

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

127) -> None: 

128 """ 

129 Delete a new Relationship between two organisations 

130 """ 

131 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

132 reltype_id = relationship_doc.reltype_id 

133 rt = user.organisation.relationship_types.filter( 

134 RelationshipType.id == reltype_id 

135 ).one() 

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

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

138 edge = ( 

139 session.query(Edge) 

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

141 .one() 

142 ) 

143 session.delete(edge) 

144 

145 

146class StandardRelTypes(str, Enum): 

147 CONSULTING_ENGAGEMENT = "Consults For" 

148 CONSULTING_PARTNERSHIP = "Partners With" 

149 SUPPLIES = "Supplies" 

150 VENDOR_EVALUATION = "Evaluates" 

151 

152 

153@http 

154def post_network_project( 

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

156) -> List[serial.NetworkRelationship]: 

157 """ 

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

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

160 """ 

161 

162 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

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

164 raise BusinessRuleViolation( 

165 "Action only permitted for Consultant Organisations" 

166 ) 

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

168 authorise.check(user, action=perms.PROJECT_ACCESS, project=project) 

169 org = user.organisation 

170 

171 rt_lookup = dict() 

172 for SRT in StandardRelTypes: 

173 try: 

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

175 except NoResultFound: 

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

177 session.add(rel) 

178 rt_lookup[SRT] = rel 

179 

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

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

182 

183 for participant in project.participants: 

184 if participant.organisation is org: 

185 continue 

186 if participant.organisation.is_consultant: 

187 rel = rt_lookup[StandardRelTypes.CONSULTING_PARTNERSHIP] 

188 else: 

189 rel = rt_lookup[StandardRelTypes.CONSULTING_ENGAGEMENT] 

190 edge = Edge( 

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

192 ) 

193 session.merge(edge) 

194 

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

196 

197 for respondent_id in respondent_id_set: 

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

199 continue 

200 

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

202 

203 rel = rt_lookup[StandardRelTypes.SUPPLIES] 

204 session.merge( 

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

206 ) 

207 

208 evaluates_rel = rt_lookup[StandardRelTypes.VENDOR_EVALUATION] 

209 session.merge( 

210 Edge( 

211 from_org_id=org.id, 

212 to_org_id=respondent_id, 

213 relationship_id=evaluates_rel.id, 

214 ) 

215 ) 

216 

217 return get_network(session, user)