Coverage for postrfp/shared/fetch/projq.py: 100%

1""" 

2Project related queries 

3""" 

4 

5import logging 

6from typing import ( 

7 List, 

8 Optional, 

9 Dict, 

10 Any, 

11) 

12 

13from sqlalchemy import ( 

14 select, 

15 or_, 

16 and_, 

17) 

18from sqlalchemy.orm import ( 

19 lazyload, 

20 Bundle, 

21 Query, 

22 Session, 

23 undefer, 

24) 

25from sqlalchemy.orm.session import object_session 

26 

27from postrfp.model import ( 

28 Participant, 

29 ProjectPermission, 

30 ProjectWatchList, 

31 ProjectNote, 

32 Organisation, 

33 Project, 

34 User, 

35 Issue, 

36 Category, 

37 Status, 

38) 

39 

40log = logging.getLogger(__name__) 

41 

42 

43def project( 

44 session: Session, 

45 project_id: int, 

46 with_description: bool = False, 

47) -> Project: 

48 """ 

49 Fetch a Project from the database 

50 

51 Parameters 

52 ---------- 

53 session : Session 

54 

55 project_id : int 

56 

57 Raises 

58 ------ 

59 NoResultFound if no Project found for the given project_id 

60 """ 

61 q = ( 

62 session.query(Project) 

63 .filter(Project.id == project_id) 

64 .options(lazyload(Project.owner_org)) 

65 ) 

66 if with_description: 

67 q.options(undefer(Project.description)) 

68 return q.one() 

69 

70 

71pw_cols: Bundle = Bundle( 

72 "listed_project", 

73 Project.id, 

74 Project.workflow_id, 

75 Project.title, 

76 Project.deadline, 

77 Organisation.name.label("owner_org_name"), 

78 Status.name.label("status"), 

79 Organisation.id.label("owner_org_id"), 

80 Project.date_created, 

81 ProjectWatchList.id.isnot(None).label("is_watched"), 

82 ProjectWatchList.date_created.label("watching_since"), 

83 single_entity=True, 

84) 

85 

86 

87def projects_with_watched( 

88 session: Session, user: User, participant_id: Optional[str] = None 

89) -> Query: 

90 org_id = user.org_id 

91 if participant_id: 

92 org_id = participant_id 

93 pq = ( 

94 session.query(pw_cols) 

95 .join(Participant) 

96 .join(Status, Project.current_status_id == Status.id) 

97 .filter(Participant.org_id == org_id) 

98 .outerjoin( 

99 ProjectWatchList, 

100 and_( 

101 ProjectWatchList.user == user, ProjectWatchList.project_id == Project.id 

102 ), 

103 ) 

104 .outerjoin(Organisation, Project.org_id == Organisation.id) 

105 ) 

106 if user.is_restricted: 

107 pq = pq.join(ProjectPermission).filter(ProjectPermission.user == user) 

108 return pq 

109 

110 

111def participant_notes_query(project: Project) -> Query: 

112 session = object_session(project) 

113 assert session is not None 

114 pq = session.query(Participant.org_id).filter(Participant.project == project) 

115 pn = project.notes_query.filter( 

116 or_(ProjectNote.private == 0, ProjectNote.org_id.in_(pq)) 

117 ).order_by(ProjectNote.note_time.desc()) 

118 return pn 

119 

120 

121def vendor_notes(issue: Issue, user: User) -> Query: 

122 nq = issue.project.notes_query 

123 own_org_id = user.org_id 

124 return nq.filter( 

125 or_( 

126 and_( 

127 ProjectNote.private == True, # noqa 

128 ProjectNote.org_id == own_org_id, 

129 ProjectNote.kind 

130 == "RespondentNote", # Posted by this vendor to himself 

131 ), 

132 and_( 

133 ProjectNote.private == False, # noqa: E712 

134 ProjectNote.kind == "IssuerNote", 

135 or_( 

136 ProjectNote.target_org_id == None, # noqa: E711 # Sent by issuer to all vendors 

137 ProjectNote.target_org_id == own_org_id, # Sent to just this vendor 

138 ), 

139 ), 

140 and_( 

141 ProjectNote.private == False, # noqa: E712 

142 ProjectNote.org_id == own_org_id, 

143 ProjectNote.kind 

144 == "RespondentNote", # Posted by this vendor to the issue 

145 ), 

146 ) 

147 ).order_by(ProjectNote.note_time.desc()) 

148 

149 

150def note(project: Project, note_id: int) -> ProjectNote: 

151 """ 

152 Fetch a ProjectNote by id checking it belongs to the given project but 

153 without checking user visibility 

154 

155 @raises NoResultFound 

156 """ 

157 return project.notes_query.filter(ProjectNote.id == note_id).one() 

158 

159 

160def issues_for_respondent( 

161 session: Session, buyer_user: User, respondent_id: str 

162) -> List[Dict[str, Any]]: 

163 """ 

164 Fetch a list of Issue/Project records for the given respondent_id 

165 which are visible to the given user 

166 """ 

167 cols = ( 

168 Issue.id.label("issue_id"), 

169 Issue.id.label("id"), 

170 Issue.status, 

171 Issue.issue_date, 

172 Issue.accepted_date, 

173 Issue.submitted_date, 

174 Issue.deadline, 

175 Issue.winloss_exposed, 

176 Issue.winloss_expiry, 

177 Project.id.label("project_id"), 

178 Project.title.label("project_title"), 

179 Project.org_id.label("project_owner"), 

180 ) 

181 q = ( 

182 session.query(*cols) 

183 .join(Project) 

184 .join(Participant) 

185 .filter( 

186 Participant.organisation == buyer_user.organisation, 

187 Issue.respondent_id == respondent_id, 

188 ) 

189 .order_by(Issue.issue_date.desc()) 

190 ) 

191 return q.all() 

192 

193 

194def category_for_user(session: Session, user: User, category_id: int) -> Category: 

195 """ 

196 Fetch the category with given id checking that it belongs to the user's organisation 

197 Returns NoResultFound if category id or user's org_id are wrong 

198 """ 

199 assert session is not None 

200 q = ( 

201 select(Category) 

202 .where(Category.id == category_id) 

203 .where(Category.organisation == user.organisation) 

204 ) 

205 return session.execute(q).scalar_one()