Coverage for app/logic/minor.py: 97%

89 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-07-31 16:31 +0000

1from collections import defaultdict 

2from typing import List, Dict 

3from playhouse.shortcuts import model_to_dict 

4from peewee import JOIN, fn, Case, DoesNotExist 

5 

6from app.models.user import User 

7from app.models.term import Term 

8from app.models.event import Event 

9from app.models.course import Course 

10from app.models.program import Program 

11from app.models.certification import Certification 

12from app.models.courseInstructor import CourseInstructor 

13from app.models.eventParticipant import EventParticipant 

14from app.models.courseParticipant import CourseParticipant 

15from app.models.individualRequirement import IndividualRequirement 

16from app.models.certificationRequirement import CertificationRequirement 

17from app.models.communityEngagementRequest import CommunityEngagementRequest 

18 

19def getEngagementTotal(engagementData): 

20 """  

21 Count the number of engagements (from all terms) that have matched with a requirement  

22 """ 

23 

24 # map the flattened list of engagements to their matched values, and sum them 

25 return sum(map(lambda e: e['matched'], sum(engagementData.values(),[]))) 

26 

27 

28def getMinorInterest() -> List[Dict]: 

29 """ 

30 Get all students that have indicated interest in the CCE minor and return a list of dicts of all interested students 

31 """ 

32 interestedStudents = (User.select(User) 

33 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(User.username == IndividualRequirement.username)) 

34 .where(User.isStudent & User.minorInterest & IndividualRequirement.username.is_null(True))) 

35 

36 interestedStudentList = [model_to_dict(student) for student in interestedStudents] 

37 

38 return interestedStudentList 

39 

40def getMinorProgress(): 

41 """ 

42 Get all the users who have an IndividualRequirement record under the CCE certification which  

43 and returns a list of dicts containing the student, how many engagements they have completed,  

44 and if they have completed the summer experience.  

45 """ 

46 summerCase = Case(None, [(CertificationRequirement.name == "Summer Program", 1)], 0) 

47 

48 engagedStudentsWithCount = ( 

49 User.select(User, fn.COUNT(IndividualRequirement.id).alias('engagementCount'), 

50 fn.SUM(summerCase).alias('hasSummer'), 

51 fn.IF(fn.COUNT(CommunityEngagementRequest.id) > 0, True, False).alias('hasCommunityEngagementRequest')) 

52 .join(IndividualRequirement, on=(User.username == IndividualRequirement.username)) 

53 .join(CertificationRequirement, on=(IndividualRequirement.requirement_id == CertificationRequirement.id)) 

54 .switch(User).join(CommunityEngagementRequest, JOIN.LEFT_OUTER, on= (User.username == CommunityEngagementRequest.user,)) 

55 .where(CertificationRequirement.certification_id == Certification.CCE) 

56 .group_by(User.firstName, User.lastName, User.username) 

57 .order_by(fn.COUNT(IndividualRequirement.id).desc()) 

58 ) 

59 engagedStudentsList = [{'username': student.username, 

60 'firstName': student.firstName, 

61 'lastName': student.lastName, 

62 'engagementCount': student.engagementCount - student.hasSummer, 

63 'hasCommunityEngagementRequest': student.hasCommunityEngagementRequest, 

64 'hasSummer': "Completed" if student.hasSummer else "Incomplete"} for student in engagedStudentsWithCount] 

65 return engagedStudentsList 

66 

67def toggleMinorInterest(username): 

68 """ 

69 Given a username, update their minor interest and minor status. 

70 """ 

71 user = User.get(username=username) 

72 user.minorInterest = not user.minorInterest 

73 

74 user.save() 

75 

76def getCourseInformation(id): 

77 """ 

78 Given a course ID, return an object containing the course information and  

79 its instructors full names. 

80 """ 

81 # retrieve the course and the course instructors 

82 course = model_to_dict(Course.get_by_id(id)) 

83 

84 courseInstructors = (CourseInstructor.select(CourseInstructor, User) 

85 .join(Course).switch() 

86 .join(User) 

87 .where(Course.id == id)) 

88 

89 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course} 

90 

91 return courseInformation 

92 

93def getProgramEngagementHistory(program_id, username, term_id): 

94 """ 

95 Given a program_id, username, and term_id, return an object containing all events in the provided program  

96 and in the given term along with the program name. 

97 """ 

98 # execute a query that will retrieve all events in which the user has participated 

99 # that fall under the provided term and programs. 

100 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, fn.SUM(EventParticipant.hoursEarned).alias("hoursEarned")) 

101 .join(Program).switch() 

102 .join(EventParticipant) 

103 .where(EventParticipant.user == username, 

104 Event.term == term_id, 

105 Program.id == program_id) 

106 ) 

107 

108 program = Program.get_by_id(program_id) 

109 

110 # calculate total amount of hours for the whole program that term 

111 totalHours = 0 

112 for event in eventsInProgramAndTerm: 

113 if event.hoursEarned: 

114 totalHours += event.hoursEarned 

115 

116 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours} 

117 

118 return participatedEvents 

119 

120def setCommunityEngagementForUser(action, engagementData, currentUser): 

121 """ 

122 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement 

123 

124 :param action: The behavior of the function. Can be 'add' or 'remove' 

125 :param engagementData: 

126 type: program or course 

127 id: program or course id 

128 username: the username of the student that is having a community engagement added or removed 

129 term: The term the engagement is recorded in 

130 :param currentuser: The user who is performing the add/remove action  

131 

132 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement 

133 """ 

134 if engagementData['type'] not in ['program','course']: 

135 raise Exception("Invalid engagement type!") 

136 

137 requirement = (CertificationRequirement.select() 

138 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=( 

139 (IndividualRequirement.requirement == CertificationRequirement.id) & 

140 (IndividualRequirement.username == engagementData['username']))) 

141 .where(IndividualRequirement.username.is_null(True), 

142 CertificationRequirement.certification == Certification.CCE, 

143 CertificationRequirement.name.not_in(['Summer Program']))) 

144 if action == 'add': 

145 try: 

146 IndividualRequirement.create(**{engagementData['type']: engagementData['id'], 

147 "username": engagementData['username'], 

148 "term": engagementData['term'], 

149 "requirement": requirement.get(), 

150 "addedBy": currentUser, 

151 }) 

152 # Thrown if there are no available engagement requirements left. Handled elsewhere. 

153 except DoesNotExist as e: 

154 raise e 

155 

156 elif action == 'remove': 

157 IndividualRequirement.delete().where( 

158 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'], 

159 IndividualRequirement.username == engagementData['username'], 

160 IndividualRequirement.term == engagementData['term'] 

161 ).execute() 

162 else: 

163 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser") 

164 

165def getCommunityEngagementByTerm(username): 

166 """ 

167 Given a username, return all of their community engagements (service learning courses and event participations.) 

168 """ 

169 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1) 

170 

171 courses = (Course.select(Course, courseMatchCase.alias("matchedReq")) 

172 .join(CourseParticipant, on=(Course.id == CourseParticipant.course)) 

173 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=( 

174 (IndividualRequirement.course == Course.id) & 

175 (IndividualRequirement.username == CourseParticipant.user) & 

176 (IndividualRequirement.term == Course.term))) 

177 .where(CourseParticipant.user == username) 

178 .group_by(Course.courseName, Course.term)) 

179 

180 # initialize default dict to store term descriptions as keys mapping to each 

181 # engagement's respective type, name, id, and term. 

182 communityEngagementByTermDict = defaultdict(list) 

183 for course in courses: 

184 communityEngagementByTermDict[(course.term.description, course.term.id)].append( 

185 {"name":course.courseName, 

186 "id":course.id, 

187 "type":"course", 

188 "matched": course.matchedReq, 

189 "term":course.term.id}) 

190 

191 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1) 

192 

193 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq')) 

194 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch() 

195 .join(Program) 

196 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) & 

197 (IndividualRequirement.username == EventParticipant.user) & 

198 (IndividualRequirement.term == Event.term))) 

199 .where(EventParticipant.user == username) 

200 .group_by(Event.program, Event.term)) 

201 

202 for event in events: 

203 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName, 

204 "id":event.program.id, 

205 "type":"program", 

206 "matched": event.matchedReq, 

207 "term":event.term.id 

208 }) 

209 

210 # sorting the communityEngagementByTermDict by the term id 

211 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1])) 

212 

213def saveOtherEngagementRequest(engagementRequest): 

214 """ 

215 Create a CommunityEngagementRequest entry based off of the form data 

216 """ 

217 engagementRequest['status'] = "Pending" 

218 CommunityEngagementRequest.create(**engagementRequest) 

219 

220 

221def saveSummerExperience(username, summerExperience, currentUser): 

222 """ 

223 :param username: username of the student that the summer experience is for 

224 :param summerExperience: dict  

225 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table) 

226 selectedSummerTerm: the term description that the summer experience took place in 

227 :param currentUser: the username of the user who added the summer experience record 

228 

229 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for  

230 'Summer Program' with the contents of summerExperience.  

231 """ 

232 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program']) 

233 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute() 

234 

235 requirement = (CertificationRequirement.select() 

236 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) & 

237 (IndividualRequirement.username == username))) 

238 .where(IndividualRequirement.username.is_null(True), 

239 CertificationRequirement.certification == Certification.CCE, 

240 CertificationRequirement.name << ['Summer Program'])) 

241 

242 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm'])) 

243 

244 IndividualRequirement.create(**{"description": summerExperience['summerExperience'], 

245 "username": username, 

246 "term": summerTerm.get(), 

247 "requirement": requirement.get(), 

248 "addedBy": currentUser, 

249 }) 

250 return "" 

251 

252def getSummerExperience(username): 

253 """ 

254 Get a students summer experience to populate text box if the student has one 

255 """ 

256 summerExperience = (IndividualRequirement.select() 

257 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch() 

258 .join(Term, on=(IndividualRequirement.term == Term.id)) 

259 .where(IndividualRequirement.username == username, 

260 CertificationRequirement.certification == Certification.CCE, 

261 CertificationRequirement.name << ['Summer Program'])) 

262 if len(list(summerExperience)) == 1: 

263 return (summerExperience.get().term.description, summerExperience.get().description) 

264 

265 return (None, None) 

266 

267def removeSummerExperience(username): 

268 """ 

269 Delete IndividualRequirement table entry for 'username' 

270 """ 

271 term, summerExperienceToDelete = getSummerExperience(username) 

272 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute() 

273 

274 

275def getSummerTerms(): 

276 """ 

277 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience 

278 """ 

279 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder)) 

280 

281 return summerTerms