Coverage for app/logic/minor.py: 91%
109 statements
« prev ^ index » next coverage.py v7.2.7, created at 2025-04-03 22:28 +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 & ~User.declaredMinor & 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, isAdding):
69 """
70 Given a username, update their minor interest and minor status.
71 """
73 try:
74 user = User.get(username=username)
75 if not user:
76 return {"error": "User not found"}, 404
78 user.minorInterest = isAdding
79 user.declaredMinor = False
80 user.save()
82 except Exception as e:
83 print(f"Error updating minor interest: {e}")
84 return {"error": str(e)}, 500
86def declareMinorInterest(username):
87 """
88 Given a username, update their minor declaration
89 """
90 user = User.get_by_id(username)
92 if not user:
93 raise ValueError(f"User with username '{username}' not found.")
95 user.declaredMinor = not user.declaredMinor
97 try:
98 user.save()
99 except Exception as e:
100 raise RuntimeError(f"Failed to declare interested student: {e}")
102def getDeclaredMinorStudents():
103 """
104 Get a list of the students who have declared minor
105 """
106 declaredStudents = User.select().where(User.isStudent & User.minorInterest & User.declaredMinor)
108 interestedStudentList = [model_to_dict(student) for student in declaredStudents]
110 return interestedStudentList
112def getCourseInformation(id):
113 """
114 Given a course ID, return an object containing the course information and
115 its instructors full names.
116 """
117 # retrieve the course and the course instructors
118 course = model_to_dict(Course.get_by_id(id))
120 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
121 .join(Course).switch()
122 .join(User)
123 .where(Course.id == id))
125 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
127 return courseInformation
129def getProgramEngagementHistory(program_id, username, term_id):
130 """
131 Given a program_id, username, and term_id, return an object containing all events in the provided program
132 and in the given term along with the program name.
133 """
134 # execute a query that will retrieve all events in which the user has participated
135 # that fall under the provided term and programs.
136 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, EventParticipant.hoursEarned)
137 .join(Program).switch()
138 .join(EventParticipant)
139 .where(EventParticipant.user == username,
140 Event.term == term_id,
141 Event.isService == True,
142 Program.id == program_id)
143 )
145 program = Program.get_by_id(program_id)
147 # calculate total amount of hours for the whole program that term
148 totalHours = 0
149 for event in eventsInProgramAndTerm:
150 if event.eventparticipant.hoursEarned:
151 totalHours += event.eventparticipant.hoursEarned
153 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
155 return participatedEvents
157def setCommunityEngagementForUser(action, engagementData, currentUser):
158 """
159 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
161 :param action: The behavior of the function. Can be 'add' or 'remove'
162 :param engagementData:
163 type: program or course
164 id: program or course id
165 username: the username of the student that is having a community engagement added or removed
166 term: The term the engagement is recorded in
167 :param currentuser: The user who is performing the add/remove action
169 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
170 """
171 if engagementData['type'] not in ['program','course']:
172 raise Exception("Invalid engagement type!")
174 requirement = (CertificationRequirement.select()
175 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
176 (IndividualRequirement.requirement == CertificationRequirement.id) &
177 (IndividualRequirement.username == engagementData['username'])))
178 .where(IndividualRequirement.username.is_null(True),
179 CertificationRequirement.certification == Certification.CCE,
180 CertificationRequirement.name.not_in(['Summer Program'])))
181 if action == 'add':
182 try:
183 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
184 "username": engagementData['username'],
185 "term": engagementData['term'],
186 "requirement": requirement.get(),
187 "addedBy": currentUser,
188 })
189 # Thrown if there are no available engagement requirements left. Handled elsewhere.
190 except DoesNotExist as e:
191 raise e
193 elif action == 'remove':
194 IndividualRequirement.delete().where(
195 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
196 IndividualRequirement.username == engagementData['username'],
197 IndividualRequirement.term == engagementData['term']
198 ).execute()
199 else:
200 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
202def getCommunityEngagementByTerm(username):
203 """
204 Given a username, return all of their community engagements (service learning courses and event participations.)
205 """
206 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
208 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
209 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
210 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
211 (IndividualRequirement.course == Course.id) &
212 (IndividualRequirement.username == CourseParticipant.user) &
213 (IndividualRequirement.term == Course.term)))
214 .where(CourseParticipant.user == username)
215 .group_by(Course.courseName, Course.term))
217 # initialize default dict to store term descriptions as keys mapping to each
218 # engagement's respective type, name, id, and term.
219 communityEngagementByTermDict = defaultdict(list)
220 for course in courses:
221 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
222 {"name":course.courseName,
223 "id":course.id,
224 "type":"course",
225 "matched": course.matchedReq,
226 "term":course.term.id})
228 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
230 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
231 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
232 .join(Program)
233 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
234 (IndividualRequirement.username == EventParticipant.user) &
235 (IndividualRequirement.term == Event.term)))
236 .where(EventParticipant.user == username, Event.isService == True)
237 .group_by(Event.program, Event.term))
239 for event in events:
240 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
241 "id":event.program.id,
242 "type":"program",
243 "matched": event.matchedReq,
244 "term":event.term.id
245 })
247 # sorting the communityEngagementByTermDict by the term id
248 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
250def saveOtherEngagementRequest(engagementRequest):
251 """
252 Create a CommunityEngagementRequest entry based off of the form data
253 """
254 engagementRequest['status'] = "Pending"
255 CommunityEngagementRequest.create(**engagementRequest)
258def saveSummerExperience(username, summerExperience, currentUser):
259 """
260 :param username: username of the student that the summer experience is for
261 :param summerExperience: dict
262 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
263 selectedSummerTerm: the term description that the summer experience took place in
264 :param currentUser: the username of the user who added the summer experience record
266 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
267 'Summer Program' with the contents of summerExperience.
268 """
269 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
270 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
272 requirement = (CertificationRequirement.select()
273 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
274 (IndividualRequirement.username == username)))
275 .where(IndividualRequirement.username.is_null(True),
276 CertificationRequirement.certification == Certification.CCE,
277 CertificationRequirement.name << ['Summer Program']))
279 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
281 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
282 "username": username,
283 "term": summerTerm.get(),
284 "requirement": requirement.get(),
285 "addedBy": currentUser,
286 })
287 return ""
289def getSummerExperience(username):
290 """
291 Get a students summer experience to populate text box if the student has one
292 """
293 summerExperience = (IndividualRequirement.select()
294 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
295 .join(Term, on=(IndividualRequirement.term == Term.id))
296 .where(IndividualRequirement.username == username,
297 CertificationRequirement.certification == Certification.CCE,
298 CertificationRequirement.name << ['Summer Program']))
299 if len(list(summerExperience)) == 1:
300 return (summerExperience.get().term.description, summerExperience.get().description)
302 return (None, None)
304def removeSummerExperience(username):
305 """
306 Delete IndividualRequirement table entry for 'username'
307 """
308 term, summerExperienceToDelete = getSummerExperience(username)
309 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()
312def getSummerTerms():
313 """
314 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience
315 """
316 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder))
318 return summerTerms