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

89 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-09-18 19:56 +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 'hasGraduated': student.hasGraduated, 

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

64 'hasCommunityEngagementRequest': student.hasCommunityEngagementRequest, 

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

66 return engagedStudentsList 

67 

68def toggleMinorInterest(username): 

69 """ 

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

71 """ 

72 user = User.get(username=username) 

73 user.minorInterest = not user.minorInterest 

74 

75 user.save() 

76 

77def getCourseInformation(id): 

78 """ 

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

80 its instructors full names. 

81 """ 

82 # retrieve the course and the course instructors 

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

84 

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

86 .join(Course).switch() 

87 .join(User) 

88 .where(Course.id == id)) 

89 

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

91 

92 return courseInformation 

93 

94def getProgramEngagementHistory(program_id, username, term_id): 

95 """ 

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

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

98 """ 

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

100 # that fall under the provided term and programs. 

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

102 .join(Program).switch() 

103 .join(EventParticipant) 

104 .where(EventParticipant.user == username, 

105 Event.term == term_id, 

106 Program.id == program_id) 

107 ) 

108 

109 program = Program.get_by_id(program_id) 

110 

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

112 totalHours = 0 

113 for event in eventsInProgramAndTerm: 

114 if event.hoursEarned: 

115 totalHours += event.hoursEarned 

116 

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

118 

119 return participatedEvents 

120 

121def setCommunityEngagementForUser(action, engagementData, currentUser): 

122 """ 

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

124 

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

126 :param engagementData: 

127 type: program or course 

128 id: program or course id 

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

130 term: The term the engagement is recorded in 

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

132 

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

134 """ 

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

136 raise Exception("Invalid engagement type!") 

137 

138 requirement = (CertificationRequirement.select() 

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

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

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

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

143 CertificationRequirement.certification == Certification.CCE, 

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

145 if action == 'add': 

146 try: 

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

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

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

150 "requirement": requirement.get(), 

151 "addedBy": currentUser, 

152 }) 

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

154 except DoesNotExist as e: 

155 raise e 

156 

157 elif action == 'remove': 

158 IndividualRequirement.delete().where( 

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

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

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

162 ).execute() 

163 else: 

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

165 

166def getCommunityEngagementByTerm(username): 

167 """ 

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

169 """ 

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

171 

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

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

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

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

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

177 (IndividualRequirement.term == Course.term))) 

178 .where(CourseParticipant.user == username) 

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

180 

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

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

183 communityEngagementByTermDict = defaultdict(list) 

184 for course in courses: 

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

186 {"name":course.courseName, 

187 "id":course.id, 

188 "type":"course", 

189 "matched": course.matchedReq, 

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

191 

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

193 

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

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

196 .join(Program) 

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

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

199 (IndividualRequirement.term == Event.term))) 

200 .where(EventParticipant.user == username) 

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

202 

203 for event in events: 

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

205 "id":event.program.id, 

206 "type":"program", 

207 "matched": event.matchedReq, 

208 "term":event.term.id 

209 }) 

210 

211 # sorting the communityEngagementByTermDict by the term id 

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

213 

214def saveOtherEngagementRequest(engagementRequest): 

215 """ 

216 Create a CommunityEngagementRequest entry based off of the form data 

217 """ 

218 engagementRequest['status'] = "Pending" 

219 CommunityEngagementRequest.create(**engagementRequest) 

220 

221 

222def saveSummerExperience(username, summerExperience, currentUser): 

223 """ 

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

225 :param summerExperience: dict  

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

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

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

229 

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

231 'Summer Program' with the contents of summerExperience.  

232 """ 

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

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

235 

236 requirement = (CertificationRequirement.select() 

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

238 (IndividualRequirement.username == username))) 

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

240 CertificationRequirement.certification == Certification.CCE, 

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

242 

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

244 

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

246 "username": username, 

247 "term": summerTerm.get(), 

248 "requirement": requirement.get(), 

249 "addedBy": currentUser, 

250 }) 

251 return "" 

252 

253def getSummerExperience(username): 

254 """ 

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

256 """ 

257 summerExperience = (IndividualRequirement.select() 

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

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

260 .where(IndividualRequirement.username == username, 

261 CertificationRequirement.certification == Certification.CCE, 

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

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

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

265 

266 return (None, None) 

267 

268def removeSummerExperience(username): 

269 """ 

270 Delete IndividualRequirement table entry for 'username' 

271 """ 

272 term, summerExperienceToDelete = getSummerExperience(username) 

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

274 

275 

276def getSummerTerms(): 

277 """ 

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

279 """ 

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

281 

282 return summerTerms