Coverage for app/logic/minor.py: 97%
88 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-04-24 15:19 +0000
1from collections import defaultdict
2from playhouse.shortcuts import model_to_dict
3from peewee import JOIN, fn, Case, DoesNotExist
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
18def getEngagementTotal(engagementData):
19 """
20 Count the number of engagements (from all terms) that have matched with a requirement
21 """
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(),[])))
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)))
35 interestedStudentList = [{'firstName': student.firstName, 'lastName': student.lastName, 'username': student.username} for student in interestedStudents]
37 return interestedStudentList
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)
47 engagedStudentsWithCount = (
48 User.select(User, fn.COUNT(IndividualRequirement.id).alias('engagementCount'),
49 fn.SUM(summerCase).alias('hasSummer'),
50 fn.IF(fn.COUNT(CommunityEngagementRequest.id) > 0, True, False).alias('hasCommunityEngagementRequest'))
51 .join(IndividualRequirement, on=(User.username == IndividualRequirement.username))
52 .join(CertificationRequirement, on=(IndividualRequirement.requirement_id == CertificationRequirement.id))
53 .switch(User).join(CommunityEngagementRequest, JOIN.LEFT_OUTER, on= (User.username == CommunityEngagementRequest.user,))
54 .where(CertificationRequirement.certification_id == Certification.CCE)
55 .group_by(User.firstName, User.lastName, User.username)
56 .order_by(fn.COUNT(IndividualRequirement.id).desc())
57 )
58 engagedStudentsList = [{'username': student.username,
59 'firstName': student.firstName,
60 'lastName': student.lastName,
61 'engagementCount': student.engagementCount - student.hasSummer,
62 'hasCommunityEngagementRequest': student.hasCommunityEngagementRequest,
63 'hasSummer': "Completed" if student.hasSummer else "Incomplete"} for student in engagedStudentsWithCount]
64 return engagedStudentsList
66def toggleMinorInterest(username):
67 """
68 Given a username, update their minor interest and minor status.
69 """
70 user = User.get(username=username)
71 user.minorInterest = not user.minorInterest
73 user.save()
75def getCourseInformation(id):
76 """
77 Given a course ID, return an object containing the course information and
78 its instructors full names.
79 """
80 # retrieve the course and the course instructors
81 course = model_to_dict(Course.get_by_id(id))
83 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
84 .join(Course).switch()
85 .join(User)
86 .where(Course.id == id))
88 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
90 return courseInformation
92def getProgramEngagementHistory(program_id, username, term_id):
93 """
94 Given a program_id, username, and term_id, return an object containing all events in the provided program
95 and in the given term along with the program name.
96 """
97 # execute a query that will retrieve all events in which the user has participated
98 # that fall under the provided term and programs.
99 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, fn.SUM(EventParticipant.hoursEarned).alias("hoursEarned"))
100 .join(Program).switch()
101 .join(EventParticipant)
102 .where(EventParticipant.user == username,
103 Event.term == term_id,
104 Program.id == program_id)
105 )
107 program = Program.get_by_id(program_id)
109 # calculate total amount of hours for the whole program that term
110 totalHours = 0
111 for event in eventsInProgramAndTerm:
112 if event.hoursEarned:
113 totalHours += event.hoursEarned
115 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
117 return participatedEvents
119def setCommunityEngagementForUser(action, engagementData, currentUser):
120 """
121 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
123 :param action: The behavior of the function. Can be 'add' or 'remove'
124 :param engagementData:
125 type: program or course
126 id: program or course id
127 username: the username of the student that is having a community engagement added or removed
128 term: The term the engagement is recorded in
129 :param currentuser: The user who is performing the add/remove action
131 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
132 """
133 if engagementData['type'] not in ['program','course']:
134 raise Exception("Invalid engagement type!")
136 requirement = (CertificationRequirement.select()
137 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
138 (IndividualRequirement.requirement == CertificationRequirement.id) &
139 (IndividualRequirement.username == engagementData['username'])))
140 .where(IndividualRequirement.username.is_null(True),
141 CertificationRequirement.certification == Certification.CCE,
142 CertificationRequirement.name.not_in(['Summer Program'])))
143 if action == 'add':
144 try:
145 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
146 "username": engagementData['username'],
147 "term": engagementData['term'],
148 "requirement": requirement.get(),
149 "addedBy": currentUser,
150 })
151 # Thrown if there are no available engagement requirements left. Handled elsewhere.
152 except DoesNotExist as e:
153 raise e
155 elif action == 'remove':
156 IndividualRequirement.delete().where(
157 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
158 IndividualRequirement.username == engagementData['username'],
159 IndividualRequirement.term == engagementData['term']
160 ).execute()
161 else:
162 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
164def getCommunityEngagementByTerm(username):
165 """
166 Given a username, return all of their community engagements (service learning courses and event participations.)
167 """
168 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
170 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
171 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
172 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
173 (IndividualRequirement.course == Course.id) &
174 (IndividualRequirement.username == CourseParticipant.user) &
175 (IndividualRequirement.term == Course.term)))
176 .where(CourseParticipant.user == username)
177 .group_by(Course.courseName, Course.term))
179 # initialize default dict to store term descriptions as keys mapping to each
180 # engagement's respective type, name, id, and term.
181 communityEngagementByTermDict = defaultdict(list)
182 for course in courses:
183 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
184 {"name":course.courseName,
185 "id":course.id,
186 "type":"course",
187 "matched": course.matchedReq,
188 "term":course.term.id})
190 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
192 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
193 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
194 .join(Program)
195 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
196 (IndividualRequirement.username == EventParticipant.user) &
197 (IndividualRequirement.term == Event.term)))
198 .where(EventParticipant.user == username)
199 .group_by(Event.program, Event.term))
201 for event in events:
202 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
203 "id":event.program.id,
204 "type":"program",
205 "matched": event.matchedReq,
206 "term":event.term.id
207 })
209 # sorting the communityEngagementByTermDict by the term id
210 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
212def saveOtherEngagementRequest(engagementRequest):
213 """
214 Create a CommunityEngagementRequest entry based off of the form data
215 """
216 engagementRequest['status'] = "Pending"
217 CommunityEngagementRequest.create(**engagementRequest)
220def saveSummerExperience(username, summerExperience, currentUser):
221 """
222 :param username: username of the student that the summer experience is for
223 :param summerExperience: dict
224 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
225 selectedSummerTerm: the term description that the summer experience took place in
226 :param currentUser: the username of the user who added the summer experience record
228 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
229 'Summer Program' with the contents of summerExperience.
230 """
231 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
232 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
234 requirement = (CertificationRequirement.select()
235 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
236 (IndividualRequirement.username == username)))
237 .where(IndividualRequirement.username.is_null(True),
238 CertificationRequirement.certification == Certification.CCE,
239 CertificationRequirement.name << ['Summer Program']))
241 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
243 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
244 "username": username,
245 "term": summerTerm.get(),
246 "requirement": requirement.get(),
247 "addedBy": currentUser,
248 })
249 return ""
251def getSummerExperience(username):
252 """
253 Get a students summer experience to populate text box if the student has one
254 """
255 summerExperience = (IndividualRequirement.select()
256 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
257 .join(Term, on=(IndividualRequirement.term == Term.id))
258 .where(IndividualRequirement.username == username,
259 CertificationRequirement.certification == Certification.CCE,
260 CertificationRequirement.name << ['Summer Program']))
261 if len(list(summerExperience)) == 1:
262 return (summerExperience.get().term.description, summerExperience.get().description)
264 return (None, None)
266def removeSummerExperience(username):
267 """
268 Delete IndividualRequirement table entry for 'username'
269 """
270 term, summerExperienceToDelete = getSummerExperience(username)
271 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()
274def getSummerTerms():
275 """
276 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience
277 """
278 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder))
280 return summerTerms