Coverage for app/logic/minor.py: 97%
89 statements
« prev ^ index » next coverage.py v7.2.7, created at 2025-03-31 20:44 +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, EventParticipant.hoursEarned)
102 .join(Program).switch()
103 .join(EventParticipant)
104 .where(EventParticipant.user == username,
105 Event.term == term_id,
106 Event.isService == True,
107 Program.id == program_id)
108 )
110 program = Program.get_by_id(program_id)
112 # calculate total amount of hours for the whole program that term
113 totalHours = 0
114 for event in eventsInProgramAndTerm:
115 if event.eventparticipant.hoursEarned:
116 totalHours += event.eventparticipant.hoursEarned
118 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
120 return participatedEvents
122def setCommunityEngagementForUser(action, engagementData, currentUser):
123 """
124 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
126 :param action: The behavior of the function. Can be 'add' or 'remove'
127 :param engagementData:
128 type: program or course
129 id: program or course id
130 username: the username of the student that is having a community engagement added or removed
131 term: The term the engagement is recorded in
132 :param currentuser: The user who is performing the add/remove action
134 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
135 """
136 if engagementData['type'] not in ['program','course']:
137 raise Exception("Invalid engagement type!")
139 requirement = (CertificationRequirement.select()
140 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
141 (IndividualRequirement.requirement == CertificationRequirement.id) &
142 (IndividualRequirement.username == engagementData['username'])))
143 .where(IndividualRequirement.username.is_null(True),
144 CertificationRequirement.certification == Certification.CCE,
145 CertificationRequirement.name.not_in(['Summer Program'])))
146 if action == 'add':
147 try:
148 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
149 "username": engagementData['username'],
150 "term": engagementData['term'],
151 "requirement": requirement.get(),
152 "addedBy": currentUser,
153 })
154 # Thrown if there are no available engagement requirements left. Handled elsewhere.
155 except DoesNotExist as e:
156 raise e
158 elif action == 'remove':
159 IndividualRequirement.delete().where(
160 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
161 IndividualRequirement.username == engagementData['username'],
162 IndividualRequirement.term == engagementData['term']
163 ).execute()
164 else:
165 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
167def getCommunityEngagementByTerm(username):
168 """
169 Given a username, return all of their community engagements (service learning courses and event participations.)
170 """
171 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
173 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
174 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
175 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
176 (IndividualRequirement.course == Course.id) &
177 (IndividualRequirement.username == CourseParticipant.user) &
178 (IndividualRequirement.term == Course.term)))
179 .where(CourseParticipant.user == username)
180 .group_by(Course.courseName, Course.term))
182 # initialize default dict to store term descriptions as keys mapping to each
183 # engagement's respective type, name, id, and term.
184 communityEngagementByTermDict = defaultdict(list)
185 for course in courses:
186 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
187 {"name":course.courseName,
188 "id":course.id,
189 "type":"course",
190 "matched": course.matchedReq,
191 "term":course.term.id})
193 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
195 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
196 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
197 .join(Program)
198 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
199 (IndividualRequirement.username == EventParticipant.user) &
200 (IndividualRequirement.term == Event.term)))
201 .where(EventParticipant.user == username, Event.isService == True)
202 .group_by(Event.program, Event.term))
204 for event in events:
205 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
206 "id":event.program.id,
207 "type":"program",
208 "matched": event.matchedReq,
209 "term":event.term.id
210 })
212 # sorting the communityEngagementByTermDict by the term id
213 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
215def saveOtherEngagementRequest(engagementRequest):
216 """
217 Create a CommunityEngagementRequest entry based off of the form data
218 """
219 engagementRequest['status'] = "Pending"
220 CommunityEngagementRequest.create(**engagementRequest)
223def saveSummerExperience(username, summerExperience, currentUser):
224 """
225 :param username: username of the student that the summer experience is for
226 :param summerExperience: dict
227 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
228 selectedSummerTerm: the term description that the summer experience took place in
229 :param currentUser: the username of the user who added the summer experience record
231 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
232 'Summer Program' with the contents of summerExperience.
233 """
234 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
235 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
237 requirement = (CertificationRequirement.select()
238 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
239 (IndividualRequirement.username == username)))
240 .where(IndividualRequirement.username.is_null(True),
241 CertificationRequirement.certification == Certification.CCE,
242 CertificationRequirement.name << ['Summer Program']))
244 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
246 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
247 "username": username,
248 "term": summerTerm.get(),
249 "requirement": requirement.get(),
250 "addedBy": currentUser,
251 })
252 return ""
254def getSummerExperience(username):
255 """
256 Get a students summer experience to populate text box if the student has one
257 """
258 summerExperience = (IndividualRequirement.select()
259 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
260 .join(Term, on=(IndividualRequirement.term == Term.id))
261 .where(IndividualRequirement.username == username,
262 CertificationRequirement.certification == Certification.CCE,
263 CertificationRequirement.name << ['Summer Program']))
264 if len(list(summerExperience)) == 1:
265 return (summerExperience.get().term.description, summerExperience.get().description)
267 return (None, None)
269def removeSummerExperience(username):
270 """
271 Delete IndividualRequirement table entry for 'username'
272 """
273 term, summerExperienceToDelete = getSummerExperience(username)
274 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()
277def getSummerTerms():
278 """
279 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience
280 """
281 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder))
283 return summerTerms