Coverage for app/logic/minor.py: 97%
89 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-07-24 12:19 +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 'engagementCount': student.engagementCount - student.hasSummer,
63 'hasCommunityEngagementRequest': student.hasCommunityEngagementRequest,
64 'hasSummer': "Completed" if student.hasSummer else "Incomplete"} for student in engagedStudentsWithCount]
65 return engagedStudentsList
67def toggleMinorInterest(username):
68 """
69 Given a username, update their minor interest and minor status.
70 """
71 user = User.get(username=username)
72 user.minorInterest = not user.minorInterest
74 user.save()
76def getCourseInformation(id):
77 """
78 Given a course ID, return an object containing the course information and
79 its instructors full names.
80 """
81 # retrieve the course and the course instructors
82 course = model_to_dict(Course.get_by_id(id))
84 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
85 .join(Course).switch()
86 .join(User)
87 .where(Course.id == id))
89 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
91 return courseInformation
93def getProgramEngagementHistory(program_id, username, term_id):
94 """
95 Given a program_id, username, and term_id, return an object containing all events in the provided program
96 and in the given term along with the program name.
97 """
98 # execute a query that will retrieve all events in which the user has participated
99 # that fall under the provided term and programs.
100 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, fn.SUM(EventParticipant.hoursEarned).alias("hoursEarned"))
101 .join(Program).switch()
102 .join(EventParticipant)
103 .where(EventParticipant.user == username,
104 Event.term == term_id,
105 Program.id == program_id)
106 )
108 program = Program.get_by_id(program_id)
110 # calculate total amount of hours for the whole program that term
111 totalHours = 0
112 for event in eventsInProgramAndTerm:
113 if event.hoursEarned:
114 totalHours += event.hoursEarned
116 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
118 return participatedEvents
120def setCommunityEngagementForUser(action, engagementData, currentUser):
121 """
122 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
124 :param action: The behavior of the function. Can be 'add' or 'remove'
125 :param engagementData:
126 type: program or course
127 id: program or course id
128 username: the username of the student that is having a community engagement added or removed
129 term: The term the engagement is recorded in
130 :param currentuser: The user who is performing the add/remove action
132 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
133 """
134 if engagementData['type'] not in ['program','course']:
135 raise Exception("Invalid engagement type!")
137 requirement = (CertificationRequirement.select()
138 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
139 (IndividualRequirement.requirement == CertificationRequirement.id) &
140 (IndividualRequirement.username == engagementData['username'])))
141 .where(IndividualRequirement.username.is_null(True),
142 CertificationRequirement.certification == Certification.CCE,
143 CertificationRequirement.name.not_in(['Summer Program'])))
144 if action == 'add':
145 try:
146 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
147 "username": engagementData['username'],
148 "term": engagementData['term'],
149 "requirement": requirement.get(),
150 "addedBy": currentUser,
151 })
152 # Thrown if there are no available engagement requirements left. Handled elsewhere.
153 except DoesNotExist as e:
154 raise e
156 elif action == 'remove':
157 IndividualRequirement.delete().where(
158 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
159 IndividualRequirement.username == engagementData['username'],
160 IndividualRequirement.term == engagementData['term']
161 ).execute()
162 else:
163 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
165def getCommunityEngagementByTerm(username):
166 """
167 Given a username, return all of their community engagements (service learning courses and event participations.)
168 """
169 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
171 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
172 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
173 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
174 (IndividualRequirement.course == Course.id) &
175 (IndividualRequirement.username == CourseParticipant.user) &
176 (IndividualRequirement.term == Course.term)))
177 .where(CourseParticipant.user == username)
178 .group_by(Course.courseName, Course.term))
180 # initialize default dict to store term descriptions as keys mapping to each
181 # engagement's respective type, name, id, and term.
182 communityEngagementByTermDict = defaultdict(list)
183 for course in courses:
184 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
185 {"name":course.courseName,
186 "id":course.id,
187 "type":"course",
188 "matched": course.matchedReq,
189 "term":course.term.id})
191 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
193 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
194 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
195 .join(Program)
196 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
197 (IndividualRequirement.username == EventParticipant.user) &
198 (IndividualRequirement.term == Event.term)))
199 .where(EventParticipant.user == username)
200 .group_by(Event.program, Event.term))
202 for event in events:
203 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
204 "id":event.program.id,
205 "type":"program",
206 "matched": event.matchedReq,
207 "term":event.term.id
208 })
210 # sorting the communityEngagementByTermDict by the term id
211 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
213def saveOtherEngagementRequest(engagementRequest):
214 """
215 Create a CommunityEngagementRequest entry based off of the form data
216 """
217 engagementRequest['status'] = "Pending"
218 CommunityEngagementRequest.create(**engagementRequest)
221def saveSummerExperience(username, summerExperience, currentUser):
222 """
223 :param username: username of the student that the summer experience is for
224 :param summerExperience: dict
225 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
226 selectedSummerTerm: the term description that the summer experience took place in
227 :param currentUser: the username of the user who added the summer experience record
229 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
230 'Summer Program' with the contents of summerExperience.
231 """
232 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
233 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
235 requirement = (CertificationRequirement.select()
236 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
237 (IndividualRequirement.username == username)))
238 .where(IndividualRequirement.username.is_null(True),
239 CertificationRequirement.certification == Certification.CCE,
240 CertificationRequirement.name << ['Summer Program']))
242 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
244 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
245 "username": username,
246 "term": summerTerm.get(),
247 "requirement": requirement.get(),
248 "addedBy": currentUser,
249 })
250 return ""
252def getSummerExperience(username):
253 """
254 Get a students summer experience to populate text box if the student has one
255 """
256 summerExperience = (IndividualRequirement.select()
257 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
258 .join(Term, on=(IndividualRequirement.term == Term.id))
259 .where(IndividualRequirement.username == username,
260 CertificationRequirement.certification == Certification.CCE,
261 CertificationRequirement.name << ['Summer Program']))
262 if len(list(summerExperience)) == 1:
263 return (summerExperience.get().term.description, summerExperience.get().description)
265 return (None, None)
267def removeSummerExperience(username):
268 """
269 Delete IndividualRequirement table entry for 'username'
270 """
271 term, summerExperienceToDelete = getSummerExperience(username)
272 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()
275def getSummerTerms():
276 """
277 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience
278 """
279 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder))
281 return summerTerms