Coverage for app/logic/minor.py: 80%
97 statements
« prev ^ index » next coverage.py v7.2.7, created at 2025-03-12 21:45 +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
30 CCEMinorProposal.create(
31 student=user,
32 proposalType = 'Summer Experience',
33 contentAreas = contentAreas,
34 status="Pending",
35 createdBy = g.current_user,
36 **formData,
37 )
38 except Exception as e:
39 print(f"Error saving summer experience: {e}")
40 raise e
42def getEngagementTotal(engagementData):
43 """
44 Count the number of engagements (from all terms) that have matched with a requirement
45 """
47 # map the flattened list of engagements to their matched values, and sum them
48 return sum(map(lambda e: e['matched'], sum(engagementData.values(),[])))
51def getMinorInterest() -> List[Dict]:
52 """
53 Get all students that have indicated interest in the CCE minor and return a list of dicts of all interested students
54 """
55 interestedStudents = (User.select(User)
56 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(User.username == IndividualRequirement.username))
57 .where(User.isStudent & User.minorInterest & IndividualRequirement.username.is_null(True)))
59 interestedStudentList = [model_to_dict(student) for student in interestedStudents]
61 return interestedStudentList
63def getMinorProgress():
64 """
65 Get all the users who have an IndividualRequirement record under the CCE certification which
66 and returns a list of dicts containing the student, how many engagements they have completed,
67 and if they have completed the summer experience.
68 """
69 summerCase = Case(None, [(CCEMinorProposal.proposalType == "Summer Experience", 1)], 0)
71 engagedStudentsWithCount = (
72 User.select(User, fn.COUNT(IndividualRequirement.id).alias('engagementCount'),
73 fn.SUM(summerCase).alias('hasSummer'),
74 fn.IF(fn.COUNT(CCEMinorProposal.id) > 0, True, False).alias('hasCCEMinorProposal'))
75 .join(IndividualRequirement, on=(User.username == IndividualRequirement.username))
76 .join(CertificationRequirement, on=(IndividualRequirement.requirement_id == CertificationRequirement.id))
77 .switch(User).join(CCEMinorProposal, JOIN.LEFT_OUTER, on= (User.username == CCEMinorProposal.student))
78 .where(CertificationRequirement.certification_id == Certification.CCE)
79 .group_by(User.firstName, User.lastName, User.username)
80 .order_by(SQL("engagementCount").desc())
81 )
82 engagedStudentsList = [{'username': student.username,
83 'firstName': student.firstName,
84 'lastName': student.lastName,
85 'hasGraduated': student.hasGraduated,
86 'engagementCount': student.engagementCount - student.hasSummer,
87 'hasCCEMinorProposal': student.hasCCEMinorProposal,
88 'hasSummer': "Completed" if student.hasSummer else "Incomplete"} for student in engagedStudentsWithCount]
89 return engagedStudentsList
91def toggleMinorInterest(username):
92 """
93 Given a username, update their minor interest and minor status.
94 """
95 user = User.get(username=username)
96 user.minorInterest = not user.minorInterest
98 user.save()
100def getCourseInformation(id):
101 """
102 Given a course ID, return an object containing the course information and
103 its instructors full names.
104 """
105 # retrieve the course and the course instructors
106 course = model_to_dict(Course.get_by_id(id))
108 courseInstructors = (CourseInstructor.select(CourseInstructor, User)
109 .join(Course).switch()
110 .join(User)
111 .where(Course.id == id))
113 courseInformation = {"instructors": [(instructor.user.firstName + " " + instructor.user.lastName) for instructor in courseInstructors], "course": course}
115 return courseInformation
117def getProgramEngagementHistory(program_id, username, term_id):
118 """
119 Given a program_id, username, and term_id, return an object containing all events in the provided program
120 and in the given term along with the program name.
121 """
122 # execute a query that will retrieve all events in which the user has participated
123 # that fall under the provided term and programs.
124 eventsInProgramAndTerm = (Event.select(Event.id, Event.name, EventParticipant.hoursEarned)
125 .join(Program).switch()
126 .join(EventParticipant)
127 .where(EventParticipant.user == username,
128 Event.term == term_id,
129 Event.isService == True,
130 Program.id == program_id)
131 )
133 program = Program.get_by_id(program_id)
135 # calculate total amount of hours for the whole program that term
136 totalHours = 0
137 for event in eventsInProgramAndTerm:
138 if event.eventparticipant.hoursEarned:
139 totalHours += event.eventparticipant.hoursEarned
141 participatedEvents = {"program":program.programName, "events": [event for event in eventsInProgramAndTerm.dicts()], "totalHours": totalHours}
143 return participatedEvents
145def setCommunityEngagementForUser(action, engagementData, currentUser):
146 """
147 Either add or remove an IndividualRequirement record for a student's Sustained Community Engagement
149 :param action: The behavior of the function. Can be 'add' or 'remove'
150 :param engagementData:
151 type: program or course
152 id: program or course id
153 username: the username of the student that is having a community engagement added or removed
154 term: The term the engagement is recorded in
155 :param currentuser: The user who is performing the add/remove action
157 :raises DoesNotExist: if there are no available CertificationRequirement slots remaining for the engagement
158 """
159 if engagementData['type'] not in ['program','course']:
160 raise Exception("Invalid engagement type!")
162 requirement = (CertificationRequirement.select()
163 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
164 (IndividualRequirement.requirement == CertificationRequirement.id) &
165 (IndividualRequirement.username == engagementData['username'])))
166 .where(IndividualRequirement.username.is_null(True),
167 CertificationRequirement.certification == Certification.CCE,
168 CertificationRequirement.name.not_in(['Summer Program'])))
169 if action == 'add':
170 try:
171 IndividualRequirement.create(**{engagementData['type']: engagementData['id'],
172 "username": engagementData['username'],
173 "term": engagementData['term'],
174 "requirement": requirement.get(),
175 "addedBy": currentUser,
176 })
177 # Thrown if there are no available engagement requirements left. Handled elsewhere.
178 except DoesNotExist as e:
179 raise e
181 elif action == 'remove':
182 IndividualRequirement.delete().where(
183 getattr(IndividualRequirement, engagementData['type']) == engagementData['id'],
184 IndividualRequirement.username == engagementData['username'],
185 IndividualRequirement.term == engagementData['term']
186 ).execute()
187 else:
188 raise Exception(f"Invalid action '{action}' sent to setCommunityEngagementForUser")
190def getCommunityEngagementByTerm(username):
191 """
192 Given a username, return all of their community engagements (service learning courses and event participations.)
193 """
194 courseMatchCase = Case(None, [(IndividualRequirement.course.is_null(True) , 0)], 1)
196 courses = (Course.select(Course, courseMatchCase.alias("matchedReq"))
197 .join(CourseParticipant, on=(Course.id == CourseParticipant.course))
198 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=(
199 (IndividualRequirement.course == Course.id) &
200 (IndividualRequirement.username == CourseParticipant.user) &
201 (IndividualRequirement.term == Course.term)))
202 .where(CourseParticipant.user == username)
203 .group_by(Course.courseName, Course.term))
205 # initialize default dict to store term descriptions as keys mapping to each
206 # engagement's respective type, name, id, and term.
207 communityEngagementByTermDict = defaultdict(list)
208 for course in courses:
209 communityEngagementByTermDict[(course.term.description, course.term.id)].append(
210 {"name":course.courseName,
211 "id":course.id,
212 "type":"course",
213 "matched": course.matchedReq,
214 "term":course.term.id})
216 programMatchCase = Case(None, [(IndividualRequirement.program.is_null(True) , 0)], 1)
218 events = (Event.select(Event, Program, programMatchCase.alias('matchedReq'))
219 .join(EventParticipant, on=(Event.id == EventParticipant.event)).switch()
220 .join(Program)
221 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.program == Program.id) &
222 (IndividualRequirement.username == EventParticipant.user) &
223 (IndividualRequirement.term == Event.term)))
224 .where(EventParticipant.user == username, Event.isService == True)
225 .group_by(Event.program, Event.term))
227 for event in events:
228 communityEngagementByTermDict[(event.term.description, event.term.id)].append({"name":event.program.programName,
229 "id":event.program.id,
230 "type":"program",
231 "matched": event.matchedReq,
232 "term":event.term.id
233 })
235 # sorting the communityEngagementByTermDict by the term id
236 return dict(sorted(communityEngagementByTermDict.items(), key=lambda engagement: engagement[0][1]))
238def createOtherEngagementRequest(username, formData):
239 """
240 Create a CCEMinorProposal entry based off of the form data
241 """
242 CCEMinorProposal.create(proposalType = 'Other Engagement',
243 createdBy = g.current_user,
244 status = 'Pending',
245 student = username,
246 **formData
247 )
250def saveSummerExperience(username, summerExperience, currentUser):
251 """
252 :param username: username of the student that the summer experience is for
253 :param summerExperience: dict
254 summerExperience: string of what the summer experience was (will be written as the 'description' in the IndividualRequirement table)
255 selectedSummerTerm: the term description that the summer experience took place in
256 :param currentUser: the username of the user who added the summer experience record
258 Delete any existing IndividualRequirement entry for 'username' if it is for 'Summer Program' and create a new IndividualRequirement entry for
259 'Summer Program' with the contents of summerExperience.
260 """
261 requirementDeleteSubSelect = CertificationRequirement.select().where(CertificationRequirement.certification == Certification.CCE, CertificationRequirement.name << ['Summer Program'])
262 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.requirement == requirementDeleteSubSelect).execute()
264 requirement = (CertificationRequirement.select()
265 .join(IndividualRequirement, JOIN.LEFT_OUTER, on=((IndividualRequirement.requirement == CertificationRequirement.id) &
266 (IndividualRequirement.username == username)))
267 .where(IndividualRequirement.username.is_null(True),
268 CertificationRequirement.certification == Certification.CCE,
269 CertificationRequirement.name << ['Summer Program']))
271 summerTerm = (Term.select().where(Term.description == summerExperience['selectedSummerTerm']))
273 IndividualRequirement.create(**{"description": summerExperience['summerExperience'],
274 "username": username,
275 "term": summerTerm.get(),
276 "requirement": requirement.get(),
277 "addedBy": currentUser,
278 })
279 return ""
281def getSummerExperience(username):
282 """
283 Get a students summer experience to populate text box if the student has one
284 """
285 summerExperience = (IndividualRequirement.select()
286 .join(CertificationRequirement, JOIN.LEFT_OUTER, on=(CertificationRequirement.id == IndividualRequirement.requirement)).switch()
287 .join(Term, on=(IndividualRequirement.term == Term.id))
288 .where(IndividualRequirement.username == username,
289 CertificationRequirement.certification == Certification.CCE,
290 CertificationRequirement.name << ['Summer Program']))
291 if len(list(summerExperience)) == 1:
292 return (summerExperience.get().term.description, summerExperience.get().description)
294 return (None, None)
296def removeSummerExperience(username):
297 """
298 Delete IndividualRequirement table entry for 'username'
299 """
300 term, summerExperienceToDelete = getSummerExperience(username)
301 IndividualRequirement.delete().where(IndividualRequirement.username == username, IndividualRequirement.description == summerExperienceToDelete).execute()
304def getSummerTerms():
305 """
306 Return a list of all terms with the isSummer flag that is marked True. Used to populate term dropdown for summer experience
307 """
308 summerTerms = list(Term.select().where(Term.isSummer).order_by(Term.termOrder))
310 return summerTerms