Coverage for app/logic/minor.py: 73%
122 statements
« prev ^ index » next coverage.py v7.2.7, created at 2025-04-08 19:13 +0000
1from collections import defaultdict
2from typing import List, Dict
3from flask import flash, g
4from playhouse.shortcuts import model_to_dict
5from peewee import JOIN, fn, Case, DoesNotExist, SQL
7from app.models.user import User
8from app.models.term import Term
9from app.models.event import Event
10from app.models.course import Course
11from app.models.program import Program
12from app.models.certification import Certification
13from app.models.courseInstructor import CourseInstructor
14from app.models.eventParticipant import EventParticipant
15from app.models.courseParticipant import CourseParticipant
16from app.models.individualRequirement import IndividualRequirement
17from app.models.certificationRequirement import CertificationRequirement
18from app.models.cceMinorProposal import CCEMinorProposal
21def createSummerExperience(username, formData):
22 """
23 Given the username of the student and the formData which includes all of
24 the SummerExperience information, create a new SummerExperience object.
25 """
26 try:
27 user = User.get(User.username == username)
28 contentAreas = ', '.join(formData.getlist('contentArea')) # Combine multiple content areas
29 CCEMinorProposal.create(
30 student=user,
31 proposalType = 'Summer Experience',
32 contentAreas = contentAreas,
33 status="Pending",
34 createdBy = g.current_user,
35 **formData,
36 )
37 except Exception as e:
38 print(f"Error saving summer experience: {e}")
39 raise e
41def getCCEMinorProposals(username):
42 proposalList = []
44 cceMinorProposals = list(CCEMinorProposal.select().where(CCEMinorProposal.student==username))
46 for experience in cceMinorProposals:
47 proposalList.append({
48 "id": experience.id,
49 "type": experience.proposalType,
50 "createdBy": experience.createdBy,
51 "supervisor": experience.supervisorName,
52 "term": experience.term,
53 "status": experience.status,
54 })
56 return proposalList
58def getEngagementTotal(engagementData):
59 """
60 Count the number of engagements (from all terms) that have matched with a requirement
61 """
63 # map the flattened list of engagements to their matched values, and sum them
64 return sum(map(lambda e: e['matched'], sum(engagementData.values(),[])))
67def getMinorInterest() -> List[Dict]:
68 """
69 Get all students that have indicated interest in the CCE minor and return a list of dicts of all interested students
70 """
71 interestedStudents = (User.select(User)
72 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(User.username == IndividualRequirement.username))
73 .where(User.isStudent & User.minorInterest & ~User.declaredMinor & IndividualRequirement.username.is_null(True)))
75 interestedStudentList = [model_to_dict(student) for student in interestedStudents]
77 return interestedStudentList
79def getMinorProgress():
80 """
81 Get all the users who have an IndividualRequirement record under the CCE certification which
82 and returns a list of dicts containing the student, how many engagements they have completed,
83 and if they have completed the summer experience.
84 """
85 summerCase = Case(None, [(CCEMinorProposal.proposalType == "Summer Experience", 1)], 0)
87 engagedStudentsWithCount = (
88 User.select(User, fn.COUNT(IndividualRequirement.id).alias('engagementCount'),
89 fn.SUM(summerCase).alias('hasSummer'),
90 fn.IF(fn.COUNT(CCEMinorProposal.id) > 0, True, False).alias('hasCCEMinorProposal'))
91 .join(IndividualRequirement, on=(User.username == IndividualRequirement.username))
92 .join(CertificationRequirement, on=(IndividualRequirement.requirement_id == CertificationRequirement.id))
93 .switch(User).join(CCEMinorProposal, JOIN.LEFT_OUTER, on= (User.username == CCEMinorProposal.student))
94 .where(CertificationRequirement.certification_id == Certification.CCE)
95 .group_by(User.firstName, User.lastName, User.username)
96 .order_by(SQL("engagementCount").desc())
97 )
98 engagedStudentsList = [{'username': student.username,
99 'firstName': student.firstName,
100 'lastName': student.lastName,
101 'hasGraduated': student.hasGraduated,
102 'engagementCount': student.engagementCount - student.hasSummer,
103 'hasCCEMinorProposal': student.hasCCEMinorProposal,
104 'hasSummer': "Completed" if student.hasSummer else "Incomplete"} for student in engagedStudentsWithCount]
105 return engagedStudentsList
107def toggleMinorInterest(username, isAdding):
108 """
109 Given a username, update their minor interest and minor status.
110 """
112 try:
113 user = User.get(username=username)
114 if not user:
115 return {"error": "User not found"}, 404
117 user.minorInterest = isAdding
118 user.declaredMinor = False
119 user.save()
121 except Exception as e:
122 print(f"Error updating minor interest: {e}")
123 return {"error": str(e)}, 500
125def declareMinorInterest(username):
126 """
127 Given a username, update their minor declaration
128 """
129 user = User.get_by_id(username)
131 if not user:
132 raise ValueError(f"User with username '{username}' not found.")
134 user.declaredMinor = not user.declaredMinor
136 try:
137 user.save()
138 except Exception as e:
139 raise RuntimeError(f"Failed to declare interested student: {e}")
141def getDeclaredMinorStudents():
142 """
143 Get a list of the students who have declared minor
144 """
145 declaredStudents = User.select().where(User.isStudent & User.minorInterest & User.declaredMinor)
147 interestedStudentList = [model_to_dict(student) for student in declaredStudents]
149 return interestedStudentList
151def getCourseInformation(id):
152 """
153 Given a course ID, return an object containing the course information and
154 its instructors full names.
155 """
156 # retrieve the course and the course instructors
157 course = model_to_dict(Course.get_by_id(id))
159 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
160 .join(Course).switch()
161 .join(User)
162 .where(Course.id == id))
164 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
166 return courseInformation
168def getProgramEngagementHistory(program_id, username, term_id):
169 """
170 Given a program_id, username, and term_id, return an object containing all events in the provided program
171 and in the given term along with the program name.
172 """
173 # execute a query that will retrieve all events in which the user has participated
174 # that fall under the provided term and programs.
175 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, EventParticipant.hoursEarned)
176 .join(Program).switch()
177 .join(EventParticipant)
178 .where(EventParticipant.user == username,
179 Event.term == term_id,
180 Event.isService == True,
181 Program.id == program_id)
182 )
184 program = Program.get_by_id(program_id)
186 # calculate total amount of hours for the whole program that term
187 totalHours = 0
188 for event in eventsInProgramAndTerm:
189 if event.eventparticipant.hoursEarned:
190 totalHours += event.eventparticipant.hoursEarned
192 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
194 return participatedEvents
196def setCommunityEngagementForUser(action, engagementData, currentUser):
197 """
198 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
200 :param action: The behavior of the function. Can be 'add' or 'remove'
201 :param engagementData:
202 type: program or course
203 id: program or course id
204 username: the username of the student that is having a community engagement added or removed
205 term: The term the engagement is recorded in
206 :param currentuser: The user who is performing the add/remove action
208 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
209 """
210 if engagementData['type'] not in ['program','course']:
211 raise Exception("Invalid engagement type!")
213 requirement = (CertificationRequirement.select()
214 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
215 (IndividualRequirement.requirement == CertificationRequirement.id) &
216 (IndividualRequirement.username == engagementData['username'])))
217 .where(IndividualRequirement.username.is_null(True),
218 CertificationRequirement.certification == Certification.CCE,
219 CertificationRequirement.name.not_in(['Summer Program'])))
220 if action == 'add':
221 try:
222 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
223 "username": engagementData['username'],
224 "term": engagementData['term'],
225 "requirement": requirement.get(),
226 "addedBy": currentUser,
227 })
228 # Thrown if there are no available engagement requirements left. Handled elsewhere.
229 except DoesNotExist as e:
230 raise e
232 elif action == 'remove':
233 IndividualRequirement.delete().where(
234 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
235 IndividualRequirement.username == engagementData['username'],
236 IndividualRequirement.term == engagementData['term']
237 ).execute()
238 else:
239 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
241def getCommunityEngagementByTerm(username):
242 """
243 Given a username, return all of their community engagements (service learning courses and event participations.)
244 """
245 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
247 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
248 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
249 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
250 (IndividualRequirement.course == Course.id) &
251 (IndividualRequirement.username == CourseParticipant.user) &
252 (IndividualRequirement.term == Course.term)))
253 .where(CourseParticipant.user == username)
254 .group_by(Course.courseName, Course.term))
256 # initialize default dict to store term descriptions as keys mapping to each
257 # engagement's respective type, name, id, and term.
258 communityEngagementByTermDict = defaultdict(list)
259 for course in courses:
260 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
261 {"name":course.courseName,
262 "id":course.id,
263 "type":"course",
264 "matched": course.matchedReq,
265 "term":course.term.id})
267 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
269 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
270 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
271 .join(Program)
272 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
273 (IndividualRequirement.username == EventParticipant.user) &
274 (IndividualRequirement.term == Event.term)))
275 .where(EventParticipant.user == username, Event.isService == True)
276 .group_by(Event.program, Event.term))
278 for event in events:
279 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
280 "id":event.program.id,
281 "type":"program",
282 "matched": event.matchedReq,
283 "term":event.term.id
284 })
286 # sorting the communityEngagementByTermDict by the term id
287 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
289def createOtherEngagementRequest(username, formData):
290 """
291 Create a CCEMinorProposal entry based off of the form data
292 """
293 user = User.get(User.username == username)
295 cceObject = CCEMinorProposal.create(proposalType = 'Other Engagement',
296 createdBy = g.current_user,
297 status = 'Pending',
298 student = user,
299 **formData
300 )
302 return cceObject
304def saveSummerExperience(username, summerExperience, currentUser):
305 """
306 :param username: username of the student that the summer experience is for
307 :param summerExperience: dict
308 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
309 selectedSummerTerm: the term description that the summer experience took place in
310 :param currentUser: the username of the user who added the summer experience record
312 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
313 'Summer Program' with the contents of summerExperience.
314 """
315 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
316 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
318 requirement = (CertificationRequirement.select()
319 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
320 (IndividualRequirement.username == username)))
321 .where(IndividualRequirement.username.is_null(True),
322 CertificationRequirement.certification == Certification.CCE,
323 CertificationRequirement.name << ['Summer Program']))
325 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
327 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
328 "username": username,
329 "term": summerTerm.get(),
330 "requirement": requirement.get(),
331 "addedBy": currentUser,
332 })
333 return ""
335def getSummerExperience(username):
336 """
337 Get a students summer experience to populate text box if the student has one
338 """
339 summerExperience = (IndividualRequirement.select()
340 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
341 .join(Term, on=(IndividualRequirement.term == Term.id))
342 .where(IndividualRequirement.username == username,
343 CertificationRequirement.certification == Certification.CCE,
344 CertificationRequirement.name << ['Summer Program']))
345 if len(list(summerExperience)) == 1:
346 return (summerExperience.get().term.description, summerExperience.get().description)
348 return (None, None)
350def removeSummerExperience(username):
351 """
352 Delete IndividualRequirement table entry for 'username'
353 """
354 term, summerExperienceToDelete = getSummerExperience(username)
355 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()