Coverage for app/logic/serviceLearningCourses.py: 84%
173 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-03-20 17:02 +0000
1from flask import g
2from peewee import fn, JOIN, ModelSelect
3import re as regex
4from openpyxl import load_workbook
5from collections import defaultdict
6from typing import DefaultDict, List, Dict, Any, Union
8from app.models import mainDB
9from app.models import DoesNotExist
10from app.models.course import Course
11from app.models.user import User
12from app.models.term import Term
13from app.models.courseInstructor import CourseInstructor
14from app.models.courseParticipant import CourseParticipant
15from app.models.courseStatus import CourseStatus
16from app.models.courseQuestion import CourseQuestion
17from app.models.attachmentUpload import AttachmentUpload
18from app.models.questionNote import QuestionNote
19from app.models.note import Note
20from app.logic.createLogs import createAdminLog
21from app.logic.fileHandler import FileHandler
22from app.logic.term import addPastTerm
24def getSLProposalInfoForUser(user: User) -> Dict[int, Dict[str, Any]]:
25 """
26 Given a user, returns a nested dictionary containing each course
27 associated with the user and its course data. Used to populate
28 Service-Learning proposal table.
29 """
30 courses: List[Course] = list(Course.select(Course, Term, User, CourseStatus)
31 .join(CourseInstructor).switch()
32 .join(Term).switch()
33 .join(CourseStatus).switch()
34 .join(User)
35 .where((CourseInstructor.user==user) | (Course.createdBy==user))
36 .order_by(Course.term.desc(), Course.status))
38 courseDict: Dict[int, Dict[str, Any]] = {}
39 for course in courses:
40 courseInstructors: List[CourseInstructor] = list(CourseInstructor.select(CourseInstructor, User).join(User).where(CourseInstructor.course==course))
41 faculty: List[str] = [f"{instructor.user.firstName} {instructor.user.lastName}" for instructor in courseInstructors]
44 courseDict[course.id] = {"id":course.id,
45 "creator":f"{course.createdBy.firstName} {course.createdBy.lastName}",
46 "name":course.courseName,
47 "faculty": faculty,
48 "term": course.term,
49 "status": course.status.status}
50 return courseDict
52def saveCourseParticipantsToDatabase(cpPreview: Dict[str, Dict[str, Dict[str, List[Dict[str, Any]]]]]) -> None:
53 for term, terminfo in cpPreview.items():
54 termObj: Term = Term.get_or_none(description = term) or addPastTerm(term)
55 if not termObj:
56 print(f"Unable to find or create term {term}")
57 continue
59 for course, courseInfo in terminfo['courses'].items():
60 if 'errorMsg' in courseInfo and courseInfo['errorMsg']:
61 print(f"Unable to save course {course}. {courseInfo['errorMsg']}")
62 continue
64 courseObj: Course = Course.get_or_create(
65 courseAbbreviation = course,
66 term = termObj,
67 defaults = {"CourseName" : "",
68 "sectionDesignation" : "",
69 "courseCredit" : "1",
70 "term" : termObj,
71 "status" : 4,
72 "createdBy" : g.current_user,
73 "serviceLearningDesignatedSections" : "",
74 "previouslyApprovedDescription" : "" })[0]
76 for userDict in courseInfo['students']:
77 if userDict['errorMsg']:
78 print(f"Unable to save student. {userDict['errorMsg']}")
79 continue
81 CourseParticipant.get_or_create(user=userDict['user'],
82 course=courseObj,
83 hoursEarned=20)
85def unapprovedCourses(termId: int) -> List[Course]:
86 """
87 Queries the database to get all the neccessary information for
88 submitted/unapproved courses.
89 """
91 unapprovedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
92 .join(CourseInstructor, JOIN.LEFT_OUTER)
93 .join(User, JOIN.LEFT_OUTER).switch(Course)
94 .join(CourseStatus).switch(Course)
95 .join(Term)
96 .where(Term.id == termId,
97 Course.status.in_([CourseStatus.SUBMITTED, CourseStatus.IN_PROGRESS]))
98 .group_by(Course, Term, CourseStatus)
99 .order_by(Course.status))
101 return unapprovedCourses
103def approvedCourses(termId: int) -> List[Course]:
104 """
105 Queries the database to get all the necessary information for
106 approved courses.
107 """
108 approvedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
109 .join(CourseInstructor, JOIN.LEFT_OUTER)
110 .join(User, JOIN.LEFT_OUTER).switch(Course)
111 .join(CourseStatus).switch(Course)
112 .join(Term)
113 .where(Term.id == termId, Course.status == CourseStatus.APPROVED)
114 .group_by(Course, Term, CourseStatus))
116 return approvedCourses
118def getInstructorCourses() -> Dict[User, str]:
119 """
120 This function queries all of the course instructors and their classes and maps
121 each instructor to its respective courses.
122 """
123 instructors: List[CourseInstructor] = list(CourseInstructor.select(CourseInstructor, User, Course)
124 .join(User).switch()
125 .join(Course))
126 instructorToCoursesMap: DefaultDict[User, str] = defaultdict(list)
128 for instructor in instructors:
129 if instructor.course.courseName not in instructorToCoursesMap[instructor.user]:
130 instructorToCoursesMap[instructor.user].append(instructor.course.courseName)
132 return dict(instructorToCoursesMap)
134########### Course Actions ###########
136def renewProposal(courseID, term) -> Course:
137 """
138 Renews proposal of ID passed in for the selected term.
139 Sets status to in progress.
140 """
141 oldCourse: Course = Course.get_by_id(courseID)
142 newCourse: Course = Course.get_by_id(courseID)
143 newCourse.id = None
144 newCourse.term = Term.get_by_id(term)
145 newCourse.status = CourseStatus.IN_PROGRESS
146 newCourse.isPreviouslyApproved = True
147 newCourse.save()
148 questions: List[CourseQuestion] = list(CourseQuestion.select().where(CourseQuestion.course==oldCourse))
149 for question in questions:
150 CourseQuestion.create(course=newCourse.id,
151 questionContent=question.questionContent,
152 questionNumber=question.questionNumber)
154 instructors = CourseInstructor.select().where(CourseInstructor.course==oldCourse.id)
155 for instructor in instructors:
156 CourseInstructor.create(course=newCourse.id,
157 user=instructor.user)
159 return newCourse
161def withdrawProposal(courseID) -> None:
162 """
163 Withdraws proposal of ID passed in. Removes foreign keys first.
164 Key Dependencies: QuestionNote, CourseQuestion, CourseParticipant,
165 CourseInstructor, Note
166 """
168 # delete syllabus
169 try:
170 syllabi: List[AttachmentUpload] = list(AttachmentUpload.select().where(AttachmentUpload.course==courseID))
171 for syllabus in syllabi:
172 FileHandler(courseId = courseID).deleteFile(syllabus.id)
174 except DoesNotExist:
175 print(f"File, {AttachmentUpload.fileName}, does not exist.")
177 # delete course object
178 course: Course = Course.get(Course.id == courseID)
179 courseName: str = course.courseName
180 questions: List[CourseQuestion] = CourseQuestion.select().where(CourseQuestion.course == course)
181 notes: List[Note] = list(Note.select(Note.id)
182 .join(QuestionNote)
183 .where(QuestionNote.question.in_(questions))
184 .distinct())
185 course.delete_instance(recursive=True)
186 for note in notes:
187 note.delete_instance()
189 createAdminLog(f"Withdrew SLC proposal: {courseName}")
191def createCourse(creator: str="No user provided") -> Course:
192 """
193 Creates and returns an empty, in progress course.
194 """
195 course: Course = Course.create(status=CourseStatus.IN_PROGRESS, createdBy=creator)
196 for number in range(1, 7):
197 CourseQuestion.create(course=course, questionNumber=number)
199 return course
201def updateCourse(courseData, attachments=None) -> Union[Course, bool]:
202 """
203 This function will take in courseData for the SLC proposal page and a dictionary
204 of instuctors assigned to the course and update the information in the db.
205 """
206 with mainDB.atomic() as transaction:
207 try:
208 course: Course = Course.get_by_id(courseData['courseID'])
210 for toggler in ["slSectionsToggle", "permanentDesignation"]:
211 courseData.setdefault(toggler, "off")
213 (Course.update(courseName=courseData["courseName"],
214 courseAbbreviation=courseData["courseAbbreviation"],
215 sectionDesignation=courseData["sectionDesignation"],
216 courseCredit=courseData["credit"],
217 isRegularlyOccurring=int(courseData["isRegularlyOccurring"]),
218 term=courseData['term'],
219 status=CourseStatus.SUBMITTED,
220 isPreviouslyApproved=int(courseData["isPreviouslyApproved"]),
221 previouslyApprovedDescription = courseData["previouslyApprovedDescription"],
222 isAllSectionsServiceLearning=("on" in courseData["slSectionsToggle"]),
223 serviceLearningDesignatedSections=courseData["slDesignation"],
224 isPermanentlyDesignated=("on" in courseData["permanentDesignation"]),
225 hasSlcComponent = int(courseData["hasSlcComponent"]))
226 .where(Course.id == course.id).execute())
228 # update the existing entry with the new question responses
229 for questionIndex in range(1, 7):
230 (CourseQuestion.update(questionContent=courseData[f"{questionIndex}"])
231 .where((CourseQuestion.questionNumber == questionIndex) &
232 (CourseQuestion.course==course)).execute())
234 # delete all course instructors and create entries for the updated instructors
235 CourseInstructor.delete().where(CourseInstructor.course == course).execute()
236 instructorList: List[str] = courseData.getlist('instructor[]')
237 for instructor in instructorList:
238 CourseInstructor.create(course=course, user=instructor)
240 # save attachments to course if applicable
241 if attachments:
242 addFile: FileHandler = FileHandler(attachments, courseId=course.id)
243 addFile.saveFiles()
245 createAdminLog(f"Saved SLC proposal: {courseData['courseName']}")
247 return Course.get_by_id(course.id)
249 except Exception as e:
250 print(e)
251 transaction.rollback()
252 return False
254########### Course Actions ###########
256def parseUploadedFile(filePath):
257 """
258 Parse an Excel document at the given `filePath` for courses and
259 course participants.
261 The return value will be a tuple. The second value is a list of
262 error message tuples. The first tuple value is the error message,
263 and the second is a 0 or 1 indicating whether the error is a
264 'general' error - 1 - or an error on a specific course, term, or
265 person. The first value is a dictionary keyed by the term
266 description. Each value is another dictionary, with a key for
267 'displayMsg' and 'errorMsg', and a 'courses' key whose value is
268 a dictionary with keys for the courses in the term. Each course
269 has a 'displayMsg' and 'errorMsg' key, and a 'students' key
270 that has a list of dictionaries with 'user', 'displayMsg', and
271 'errorMsg' keys.
272 E.g.,
273 {
274 "Fall 2021": {
275 "displayMsg": "",
276 "errorMsg": "",
277 "courses": {
278 "CSC 330": {
279 "displayMsg: "CSC 330 will be created",
280 "errorMsg: "",
281 "students": [
282 {'user':'ramsayb2',
283 'displayMsg': 'Brian Ramsay',
284 'errorMsg': ''},
285 {'user':'B0073235',
286 'displayMsg': '',
287 'errorMsg': 'ERROR: B0073235 does not exist!'}]
288 },
289 "CSC 226": {
290 "displayMsg": "CSC 226 matched to existing course 'Data Structures'.",
291 "errorMsg": "",
292 "students": [
293 {'user':'ramsayb2',
294 'displayMsg': 'Brian Ramsay',
295 'errorMsg': ''},
296 {'user':'lamichhanes',
297 'displayMsg': 'Sandesh Lamichhane',
298 'errorMsg': ''}]
299 }
300 }
301 }
302 }
303 """
304 excelData = load_workbook(filename=filePath)
305 excelSheet = excelData.active
307 result = {}
308 errors = []
309 term = ''
310 course = ''
311 cellRow = 0
313 for row in excelSheet.iter_rows():
314 cellRow += 1
315 cellVal = row[0].value
316 if not cellVal:
317 continue
319 # Look for a Term. Examples: Fall 2020 or Spring B 2021
320 if regex.search(r"\b[a-zA-Z]{3,}( [AB])? \d{4}\b", str(cellVal)):
321 errorMsg = ''
322 if "Spring A" in cellVal or "Spring B" in cellVal:
323 cellVal = "Spring " + cellVal.split()[-1]
324 if "Fall A" in cellVal or "Fall B" in cellVal:
325 cellVal = "Fall " + cellVal.split()[-1]
327 if cellVal.split()[0] not in ["Summer", "Spring", "Fall", "May"]:
328 errorMsg = f"ERROR: '{cellVal}' is not a valid term in row {cellRow}."
329 else:
330 latestTerm = Term.select().order_by(Term.termOrder.desc()).get()
331 isFutureTerm = latestTerm.termOrder < Term.convertDescriptionToTermOrder(cellVal)
332 if isFutureTerm:
333 errors.append(f"ERROR: '{cellVal}' is a future term in row {cellRow}.")
334 else:
335 validTerm = Term.get_or_none(Term.description == cellVal)
337 term = cellVal
338 result[term] = {
339 'displayMsg': term,
340 'errorMsg': errorMsg,
341 'courses': {}
342 }
343 if errorMsg:
344 errors.append((errorMsg,0))
346 # Look for a Course. Examples: FRN134 CSC 226
347 elif regex.search(r"\b[A-Z]{2,4} ?\d{3}\b", str(cellVal)):
348 errorMsg = displayMsg = ''
349 if not term:
350 displayMsg = cellVal
351 errorMsg = "ERROR: No term was given for this course"
352 else:
353 existingCourse = Course.get_or_none(Course.courseAbbreviation == cellVal)
354 displayMsg = f'{cellVal} will be created.'
355 if existingCourse:
356 displayMsg = f"{cellVal} matched to the existing course {existingCourse.courseName}."
358 course = cellVal
359 result[term]['courses'][course] = {
360 'displayMsg': displayMsg,
361 'errorMsg': errorMsg,
362 'students': []
363 }
364 if errorMsg:
365 errors.append((errorMsg,0))
367 # Look for a B-Number. Example: B00123456
368 elif regex.search(r"\b[B]\d{8}\b", str(cellVal)):
369 errorMsg = displayMsg = ''
370 if not course:
371 errorMsg = "ERROR: No course is connected to this student"
372 else:
373 existingUser = User.get_or_none(User.bnumber == cellVal)
374 if existingUser:
375 displayMsg = f"{existingUser.firstName} {existingUser.lastName}"
376 existingUser = existingUser.username
377 else:
378 errorMsg = f'ERROR: {row[1].value} with B# "{row[0].value}" does not exist.'
380 result[term]['courses'][course]['students'].append({
381 'user': (existingUser or cellVal),
382 'displayMsg': displayMsg,
383 'errorMsg': errorMsg})
384 if errorMsg:
385 errors.append((errorMsg,0))
387 elif cellVal: # but didn't match the regex
388 errors.append((f'ERROR: "{cellVal}" in row {cellRow} of the Excel document does not appear to be a term, course, or valid B#.',1))
391 return result, errors