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

88 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-03-20 17:02 +0000

1from collections import defaultdict 

2from playhouse.shortcuts import model_to_dict 

3from peewee import JOIN, fn, Case, DoesNotExist 

4 

5from app.models.user import User 

6from app.models.term import Term 

7from app.models.event import Event 

8from app.models.course import Course 

9from app.models.program import Program 

10from app.models.certification import Certification 

11from app.models.courseInstructor import CourseInstructor 

12from app.models.eventParticipant import EventParticipant 

13from app.models.courseParticipant import CourseParticipant 

14from app.models.individualRequirement import IndividualRequirement 

15from app.models.certificationRequirement import CertificationRequirement 

16from app.models.communityEngagementRequest import CommunityEngagementRequest 

17 

18def getEngagementTotal(engagementData): 

19 """  

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

21 """ 

22 

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

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

25 

26 

27def getMinorInterest(): 

28 """ 

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

30 """ 

31 interestedStudents = (User.select(User.firstName, User.lastName, User.username) 

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

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

34 

35 interestedStudentList = [{'firstName': student.firstName, 'lastName': student.lastName, 'username': student.username} for student in interestedStudents] 

36 

37 return interestedStudentList 

38 

39def getMinorProgress(): 

40 """ 

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

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

43 and if they have completed the summer experience.  

44 """ 

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

46 

47 engagedStudentsWithCount = ( 

48 User.select(User.firstName, User.lastName, User.username, fn.COUNT(IndividualRequirement.id).alias('engagementCount'), fn.SUM(summerCase).alias('hasSummer')) 

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

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

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

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

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

54 ) 

55 

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

57 'firstName': student.firstName, 

58 'lastName': student.lastName, 

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

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

61 

62 return engagedStudentsList 

63 

64def toggleMinorInterest(username): 

65 """ 

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

67 """ 

68 user = User.get(username=username) 

69 user.minorInterest = not user.minorInterest 

70 

71 user.save() 

72 

73def getCourseInformation(id): 

74 """ 

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

76 its instructors full names. 

77 """ 

78 # retrieve the course and the course instructors 

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

80 

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

82 .join(Course).switch() 

83 .join(User) 

84 .where(Course.id == id)) 

85 

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

87 

88 return courseInformation 

89 

90def getProgramEngagementHistory(program_id, username, term_id): 

91 """ 

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

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

94 """ 

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

96 # that fall under the provided term and programs. 

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

98 .join(Program).switch() 

99 .join(EventParticipant) 

100 .where(EventParticipant.user == username, 

101 Event.term == term_id, 

102 Program.id == program_id) 

103 ) 

104 

105 program = Program.get_by_id(program_id) 

106 

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

108 totalHours = 0 

109 for event in eventsInProgramAndTerm: 

110 if event.hoursEarned: 

111 totalHours += event.hoursEarned 

112 

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

114 

115 return participatedEvents 

116 

117def setCommunityEngagementForUser(action, engagementData, currentUser): 

118 """ 

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

120 

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

122 :param engagementData: 

123 type: program or course 

124 id: program or course id 

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

126 term: The term the engagement is recorded in 

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

128 

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

130 """ 

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

132 raise Exception("Invalid engagement type!") 

133 

134 requirement = (CertificationRequirement.select() 

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

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

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

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

139 CertificationRequirement.certification == Certification.CCE, 

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

141 if action == 'add': 

142 try: 

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

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

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

146 "requirement": requirement.get(), 

147 "addedBy": currentUser, 

148 }) 

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

150 except DoesNotExist as e: 

151 raise e 

152 

153 elif action == 'remove': 

154 IndividualRequirement.delete().where( 

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

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

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

158 ).execute() 

159 else: 

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

161 

162 

163def getCommunityEngagementByTerm(username): 

164 """ 

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

166 """ 

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

168 

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

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

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

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

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

174 (IndividualRequirement.term == Course.term))) 

175 .where(CourseParticipant.user == username) 

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

177 

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

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

180 communityEngagementByTermDict = defaultdict(list) 

181 for course in courses: 

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

183 {"name":course.courseName, 

184 "id":course.id, 

185 "type":"course", 

186 "matched": course.matchedReq, 

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

188 

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

190 

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

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

193 .join(Program) 

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

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

196 (IndividualRequirement.term == Event.term))) 

197 .where(EventParticipant.user == username) 

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

199 

200 for event in events: 

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

202 "id":event.program.id, 

203 "type":"program", 

204 "matched": event.matchedReq, 

205 "term":event.term.id 

206 }) 

207 

208 # sorting the communityEngagementByTermDict by the term id 

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

210 

211def saveOtherEngagementRequest(engagementRequest): 

212 """ 

213 Create a CommunityEngagementRequest entry based off of the form data 

214 """ 

215 engagementRequest['status'] = "Pending" 

216 CommunityEngagementRequest.create(**engagementRequest) 

217 

218def saveSummerExperience(username, summerExperience, currentUser): 

219 """ 

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

221 :param summerExperience: dict  

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

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

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

225 

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

227 'Summer Program' with the contents of summerExperience.  

228 """ 

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

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

231 

232 requirement = (CertificationRequirement.select() 

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

234 (IndividualRequirement.username == username))) 

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

236 CertificationRequirement.certification == Certification.CCE, 

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

238 

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

240 

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

242 "username": username, 

243 "term": summerTerm.get(), 

244 "requirement": requirement.get(), 

245 "addedBy": currentUser, 

246 }) 

247 return "" 

248 

249def getSummerExperience(username): 

250 """ 

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

252 """ 

253 summerExperience = (IndividualRequirement.select() 

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

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

256 .where(IndividualRequirement.username == username, 

257 CertificationRequirement.certification == Certification.CCE, 

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

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

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

261 

262 return (None, None) 

263 

264def removeSummerExperience(username): 

265 """ 

266 Delete IndividualRequirement table entry for 'username' 

267 """ 

268 term, summerExperienceToDelete = getSummerExperience(username) 

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

270 

271 

272def getSummerTerms(): 

273 """ 

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

275 """ 

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

277 

278 return summerTerms