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

1""" 

2Define a network of relationships between organisations 

3""" 

4from enum import Enum 

5from rfpy.model.exc import BusinessRuleViolation 

6from rfpy.model.humans import OrganisationType 

7from typing import List 

8 

9from sqlalchemy.orm import Session 

10from sqlalchemy.orm.exc import NoResultFound 

11 

12from rfpy.api import validate, fetch 

13from rfpy.auth import perms 

14from rfpy.suxint import http 

15from rfpy.model import User, Edge, RelationshipType 

16 

17from rfpy.web import serial 

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.from_orm(r).dict() 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 validate.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(synchronize_session=False) 

38 

39 

40@http 

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

42 ''' 

43 Get an array of Relationship Types for your organisation 

44 ''' 

45 return [ 

46 dict(id=rt.id, name=rt.name, description=rt.description) 

47 for rt in user.organisation.relationship_types 

48 ] 

49 

50 

51@http 

52def post_reltype(session: Session, user: User, reltype_doc: serial.RelationshipType) -> serial.Id: 

53 ''' 

54 Create a new Relationship Type for your organisation 

55 

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

57 be defined between two 3rd party organisations. 

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

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

60 'Regulator', etc. 

61 

62 @permission MANAGE_ORGANISATION 

63 ''' 

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

65 rt = RelationshipType(**reltype_doc) 

66 user.organisation.relationship_types.append(rt) 

67 session.flush() 

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

69 

70 

71@http 

72def put_reltype(session: Session, user: User, reltype_id, reltype_doc) -> serial.Id: 

73 ''' 

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

75 ''' 

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

77 rt = user.organisation.relationship_types.filter(RelationshipType.id == reltype_id).one() 

78 rt.name = reltype_doc['name'] 

79 if 'description' in reltype_doc: 

80 rt.description = reltype_doc['description'] 

81 

82 

83@http 

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

85 ''' 

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

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

88 ''' 

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

90 rt = user.organisation.relationship_types.filter(RelationshipType.id == reltype_id).one() 

91 session.delete(rt) 

92 

93 

94@http 

95def post_relationship(session: Session, user: User, relationship_doc): 

96 ''' 

97 Create a new Relationship between two organisations 

98 ''' 

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

100 reltype_id = relationship_doc['reltype_id'] 

101 rt = user.organisation.relationship_types.filter(RelationshipType.id == reltype_id).one() 

102 from_org = fetch.organisation(session, relationship_doc['from_org_id']) 

103 to_org = fetch.organisation(session, relationship_doc['to_org_id']) 

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

105 session.add(edge) 

106 

107 

108@http 

109def delete_relationship(session: Session, user: User, relationship_doc): 

110 ''' 

111 Delete a new Relationship between two organisations 

112 ''' 

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

114 reltype_id = relationship_doc['reltype_id'] 

115 rt = user.organisation.relationship_types.filter(RelationshipType.id == reltype_id).one() 

116 from_org = fetch.organisation(session, relationship_doc['from_org_id']) 

117 to_org = fetch.organisation(session, relationship_doc['to_org_id']) 

118 edge = session.query(Edge)\ 

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

120 .one() 

121 session.delete(edge) 

122 

123 

124class StandardRelTypes(str, Enum): 

125 CONSULTING_ENGAGEMENT = 'Consults For' 

126 CONSULTING_PARTNERSHIP = 'Partners With' 

127 SUPPLIES = 'Supplies' 

128 VENDOR_EVALUATION = 'Evaluates' 

129 

130 

131@http 

132def post_network_project(session: Session, 

133 user: User, 

134 project_id: int) -> List[serial.NetworkRelationship]: 

135 ''' 

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

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

138 ''' 

139 

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

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

142 raise BusinessRuleViolation('Action only permitted for Consultant Organisations') 

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

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

145 org = user.organisation 

146 

147 rt_lookup = dict() 

148 for SRT in StandardRelTypes: 

149 

150 try: 

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

152 except NoResultFound: 

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

154 session.add(rel) 

155 rt_lookup[SRT] = rel 

156 

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

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

159 

160 for participant in project.participants: 

161 if participant.organisation is org: 

162 continue 

163 if participant.organisation.is_consultant: 

164 rel = rt_lookup[StandardRelTypes.CONSULTING_PARTNERSHIP] 

165 else: 

166 rel = rt_lookup[StandardRelTypes.CONSULTING_ENGAGEMENT] 

167 edge = Edge(from_org_id=org.id, to_org_id=participant.org_id, relationship_id=rel.id) 

168 session.merge(edge) 

169 

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

171 

172 for respondent_id in respondent_id_set: 

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

174 continue 

175 

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

177 

178 rel = rt_lookup[StandardRelTypes.SUPPLIES] 

179 session.merge(Edge(from_org_id=respondent_id, to_org_id=buyer_id, relationship_id=rel.id)) 

180 

181 evaluates_rel = rt_lookup[StandardRelTypes.VENDOR_EVALUATION] 

182 session.merge(Edge(from_org_id=org.id, to_org_id=respondent_id, 

183 relationship_id=evaluates_rel.id)) 

184 

185 return get_network(session, user)