Coverage for app/logic/minor.py: 97%
89 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-09-06 19:35 +0000
1from collections import defaultdict
2from typing import List, Dict
3from playhouse.shortcuts import model_to_dict
4from peewee import JOIN, fn, Case, DoesNotExist
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
19def getEngagementTotal(engagementData):
20 """
21 Count the number of engagements (from all terms) that have matched with a requirement
22 """
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(),[])))
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)))
36 interestedStudentList = [model_to_dict(student) for student in interestedStudents]
38 return interestedStudentList
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)
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
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
75 user.save()
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))
85 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
86 .join(Course).switch()
87 .join(User)
88 .where(Course.id == id))
90 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
92 return courseInformation
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 )
109 program = Program.get_by_id(program_id)
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
117 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
119 return participatedEvents
121def setCommunityEngagementForUser(action, engagementData, currentUser):
122 """
123 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
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
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!")
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
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")
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)
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))
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})
192 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
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))
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 })
211 # sorting the communityEngagementByTermDict by the term id
212 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
214def saveOtherEngagementRequest(engagementRequest):
215 """
216 Create a CommunityEngagementRequest entry based off of the form data
217 """
218 engagementRequest['status'] = "Pending"
219 CommunityEngagementRequest.create(**engagementRequest)
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
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()
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']))
243 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
245 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
246 "username": username,
247 "term": summerTerm.get(),
248 "requirement": requirement.get(),
249 "addedBy": currentUser,
250 })
251 return ""
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)
266 return (None, None)
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()
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))
282 return summerTerms