Coverage for app/logic/serviceLearningCourses.py: 80%
219 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-08-16 18:52 +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 createActivityLog
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 "abbr": course.courseAbbreviation,
48 "courseDisplayName": createCourseDisplayName(course.courseName, course.courseAbbreviation),
49 "faculty": faculty,
50 "term": course.term,
51 "status": course.status.status}
52 return courseDict
54def createCourseDisplayName(name, abbreviation):
55 '''
56 This function combines course name and numbers with conditions
57 inputs: course name, course abbreviation
58 '''
59 if name and abbreviation:
60 return f"{abbreviation} - {name}"
61 elif not name and not abbreviation:
62 return ''
63 elif not name:
64 return abbreviation
65 elif not abbreviation:
66 return name
68def saveCourseParticipantsToDatabase(cpPreview: Dict[str, Dict[str, Dict[str, List[Dict[str, Any]]]]]) -> None:
69 for term, terminfo in cpPreview.items():
70 currentCourseTerm: Term = Term.get_or_none(description = term) or addPastTerm(term)
71 if not currentCourseTerm:
72 print(f"Unable to find or create term {term}")
73 continue
75 for course, courseInfo in terminfo['courses'].items():
76 if 'errorMsg' in courseInfo and courseInfo['errorMsg']:
77 print(f"Unable to save course {course}. {courseInfo['errorMsg']}")
78 continue
80 existingCourses = list((Course.select()
81 .join(Term, on =(Course.term == Term.id))
82 .where((Course.courseAbbreviation == course) & (Term.termOrder <= currentCourseTerm.termOrder))
83 .order_by(Course.term.desc())
84 .limit(1)))
86 if not existingCourses :
88 courseObj: Course = Course.create(courseName = "",
89 sectionDesignation = "",
90 courseAbbreviation = course,
91 courseCredit = "1",
92 term = currentCourseTerm,
93 status = CourseStatus.IN_PROGRESS,
94 createdBy = g.current_user,
95 serviceLearningDesignatedSections = "",
96 previouslyApprovedDescription = "")
97 else:
98 previousMatchedCourse = existingCourses[0]
100 previousCourseTerm = Term.get(Term.id == previousMatchedCourse.term)
102 if previousCourseTerm == currentCourseTerm:
103 courseObj : Course = previousMatchedCourse
105 else:
107 courseObj: Course = Course.create(courseName = previousMatchedCourse.courseName,
108 courseAbbreviation = previousMatchedCourse.courseAbbreviation,
109 sectionDesignation = previousMatchedCourse.sectionDesignation,
110 courseCredit = previousMatchedCourse.courseCredit,
111 term = currentCourseTerm,
112 status = CourseStatus.IN_PROGRESS,
113 createdBy = g.current_user,
114 serviceLearningDesignatedSections = previousMatchedCourse.serviceLearningDesignatedSections,
115 previouslyApprovedDescription = previousMatchedCourse.previouslyApprovedDescription)
117 questions: List[CourseQuestion] = list(CourseQuestion.select().where(CourseQuestion.course == previousMatchedCourse.id))
119 for question in questions:
120 CourseQuestion.create(course = courseObj.id,
121 questionContent= question.questionContent,
122 questionNumber=question.questionNumber)
124 instructors = CourseInstructor.select().where(CourseInstructor.course == previousMatchedCourse.id)
126 for instructor in instructors:
127 CourseInstructor.create(course = courseObj.id,
128 user = instructor.user)
131 for userDict in courseInfo['students']:
132 if userDict['errorMsg']:
133 print(f"Unable to save student. {userDict['errorMsg']}")
134 continue
136 CourseParticipant.get_or_create(user=userDict['user'],
137 course=courseObj,
138 hoursEarned=20)
140def unapprovedCourses(termId: int) -> List[Course]:#
141 """
142 Queries the database to get all the neccessary information for
143 submitted/unapproved courses.
144 """
146 unapprovedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
147 .join(CourseInstructor, JOIN.LEFT_OUTER)
148 .join(User, JOIN.LEFT_OUTER).switch(Course)
149 .join(CourseStatus).switch(Course)
150 .join(Term)
151 .where(Term.id == termId,
152 Course.status.in_([CourseStatus.SUBMITTED, CourseStatus.IN_PROGRESS]))
153 .group_by(Course, Term, CourseStatus)
154 .order_by(Course.status))
156 return unapprovedCourses
158def approvedCourses(termId: int) -> List[Course]:
159 """
160 Queries the database to get all the necessary information for
161 approved courses.
162 """
163 approvedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
164 .join(CourseInstructor, JOIN.LEFT_OUTER)
165 .join(User, JOIN.LEFT_OUTER).switch(Course)
166 .join(CourseStatus).switch(Course)
167 .join(Term)
168 .where(Term.id == termId, Course.status == CourseStatus.APPROVED)
169 .group_by(Course, Term, CourseStatus))
171 return approvedCourses
173def getImportedCourses(termId: int) -> List[Course]:
174 """
175 Queries the database to get all the necessary information for
176 imported courses.
177 """
178 importedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
179 .join(CourseInstructor, JOIN.LEFT_OUTER)
180 .join(User, JOIN.LEFT_OUTER).switch(Course)
181 .join(CourseStatus).switch(Course)
182 .join(Term)
183 .where(Term.id == termId, Course.status == CourseStatus.IMPORTED)
184 .group_by(Course, Term, CourseStatus))
186 return importedCourses
188def getInstructorCourses() -> Dict[User, str]:
189 """
190 This function queries all of the course instructors and their classes and maps
191 each instructor to its respective courses.
192 """
193 instructors: List[CourseInstructor] = list(CourseInstructor.select(CourseInstructor, User, Course)
194 .join(User).switch()
195 .join(Course))
196 instructorToCoursesMap: DefaultDict[User, str] = defaultdict(list)
198 for instructor in instructors:
199 if instructor.course.courseName not in instructorToCoursesMap[instructor.user]:
200 instructorToCoursesMap[instructor.user].append(instructor.course.courseName)
202 return dict(instructorToCoursesMap)
204########### Course Actions ###########
206def renewProposal(courseID, term) -> Course:
207 """
208 Renews proposal of ID passed in for the selected term.
209 Sets status to in progress.
210 """
211 oldCourse: Course = Course.get_by_id(courseID)
212 newCourse: Course = Course.get_by_id(courseID)
213 newCourse.id = None
214 newCourse.term = Term.get_by_id(term)
215 newCourse.status = CourseStatus.IN_PROGRESS
216 newCourse.isPreviouslyApproved = True
217 newCourse.save()
218 questions: List[CourseQuestion] = list(CourseQuestion.select().where(CourseQuestion.course==oldCourse))
219 for question in questions:
220 CourseQuestion.create(course=newCourse.id,
221 questionContent=question.questionContent,
222 questionNumber=question.questionNumber)
224 instructors = CourseInstructor.select().where(CourseInstructor.course==oldCourse.id)
225 for instructor in instructors:
226 CourseInstructor.create(course=newCourse.id,
227 user=instructor.user)
229 return newCourse
231def withdrawProposal(courseID) -> None:
232 """
233 Withdraws proposal of ID passed in. Removes foreign keys first.
234 Key Dependencies: QuestionNote, CourseQuestion, CourseParticipant,
235 CourseInstructor, Note
236 """
238 # delete syllabus
239 try:
240 syllabi: List[AttachmentUpload] = list(AttachmentUpload.select().where(AttachmentUpload.course==courseID))
241 for syllabus in syllabi:
242 FileHandler(courseId = courseID).deleteFile(syllabus.id)
244 except DoesNotExist:
245 print(f"File, {AttachmentUpload.fileName}, does not exist.")
247 # deletes course
248 deletedCourse = deleteCourseObject(courseID=courseID)
250 createActivityLog(f"Withdrew SLC proposal: {deletedCourse}")
252def deleteCourseObject(courseID):
253 course: Course = Course.get(Course.id == courseID)
254 courseName: str = course.courseName
255 questions: List[CourseQuestion] = CourseQuestion.select().where(CourseQuestion.course == course)
256 notes: List[Note] = list(Note.select(Note.id)
257 .join(QuestionNote)
258 .where(QuestionNote.question.in_(questions))
259 .distinct())
260 course.delete_instance(recursive=True)
261 for note in notes:
262 note.delete_instance()
263 return courseName
265def createCourse(creator: str="No user provided") -> Course:
266 """
267 Creates and returns an empty, in progress course.
268 """
269 course: Course = Course.create(status=CourseStatus.IN_PROGRESS, createdBy=creator)
270 for number in range(1, 7):
271 CourseQuestion.create(course=course, questionNumber=number)
273 return course
275def updateCourse(courseData, attachments=None) -> Union[Course, bool]:
276 """
277 This function will take in courseData for the SLC proposal page and a dictionary
278 of instuctors assigned to the course and update the information in the db.
279 """
280 with mainDB.atomic() as transaction:
281 try:
282 course: Course = Course.get_by_id(courseData['courseID'])
284 for toggler in ["slSectionsToggle", "permanentDesignation"]:
285 courseData.setdefault(toggler, "off")
287 (Course.update(courseName=courseData["courseName"],
288 courseAbbreviation=courseData["courseAbbreviation"],
289 sectionDesignation=courseData["sectionDesignation"],
290 courseCredit=courseData["credit"],
291 isRegularlyOccurring=int(courseData["isRegularlyOccurring"]),
292 term=courseData['term'],
293 status=CourseStatus.SUBMITTED,
294 isPreviouslyApproved=int(courseData["isPreviouslyApproved"]),
295 previouslyApprovedDescription = courseData["previouslyApprovedDescription"],
296 isAllSectionsServiceLearning=("on" in courseData["slSectionsToggle"]),
297 serviceLearningDesignatedSections=courseData["slDesignation"],
298 isPermanentlyDesignated=("on" in courseData["permanentDesignation"]),
299 hasSlcComponent = int(courseData["hasSlcComponent"]))
300 .where(Course.id == course.id).execute())
302 # update the existing entry with the new question responses
303 for questionIndex in range(1, 7):
304 (CourseQuestion.update(questionContent=courseData[f"{questionIndex}"])
305 .where((CourseQuestion.questionNumber == questionIndex) &
306 (CourseQuestion.course==course)).execute())
308 # delete all course instructors and create entries for the updated instructors
309 CourseInstructor.delete().where(CourseInstructor.course == course).execute()
310 instructorList: List[str] = courseData.getlist('instructor[]')
311 for instructor in instructorList:
312 CourseInstructor.create(course=course, user=instructor)
314 # save attachments to course if applicable
315 if attachments:
316 addFile: FileHandler = FileHandler(attachments, courseId=course.id)
317 addFile.saveFiles()
319 createActivityLog(f"Saved SLC proposal: {courseData['courseName']}")
321 return Course.get_by_id(course.id)
323 except Exception as e:
324 print(e)
325 transaction.rollback()
326 return False
328def editImportedCourses(courseData):
329 """
330 This function will take in courseData for the SLC proposal page and a dictionary
331 of instructors assigned to the imported course after that one is edited
332 and update the information in the db.
333 """
335 with mainDB.atomic() as transaction:
336 try:
337 course = Course.get_by_id(courseData["courseId"])
339 Course.update(courseName=courseData["courseName"]).where(Course.id == course.id).execute()
341 (CourseParticipant.update(hoursEarned=courseData["hoursEarned"])
342 .where(CourseParticipant.course_id == course.id).execute())
344 instructorList = []
345 CourseInstructor.delete().where(CourseInstructor.course == course).execute()
347 if 'instructor[]' in courseData:
348 instructorList = courseData.getlist('instructor[]')
350 for instructor in instructorList:
351 # Checks that empty string is not added as a course instructor because some keys in the dictionary are empty string.
352 if instructor:
353 CourseInstructor.create(course=course, user=instructor)
355 return Course.get_by_id(course.id)
357 except Exception as e:
358 print(e)
359 transaction.rollback()
360 return False
362########### Course Actions ###########
364def parseUploadedFile(filePath):
365 """
366 Parse an Excel document at the given `filePath` for courses and
367 course participants.
369 The return value will be a tuple. The second value is a list of
370 error message tuples. The first tuple value is the error message,
371 and the second is a 0 or 1 indicating whether the error is a
372 'general' error - 1 - or an error on a specific course, term, or
373 person. The first value is a dictionary keyed by the term
374 description. Each value is another dictionary, with a key for
375 'displayMsg' and 'errorMsg', and a 'courses' key whose value is
376 a dictionary with keys for the courses in the term. Each course
377 has a 'displayMsg' and 'errorMsg' key, and a 'students' key
378 that has a list of dictionaries with 'user', 'displayMsg', and
379 'errorMsg' keys.
380 E.g.,
381 {
382 "Fall 2021": {
383 "displayMsg": "",
384 "errorMsg": "",
385 "courses": {
386 "CSC 330": {
387 "displayMsg: "CSC 330 will be created",
388 "errorMsg: "",
389 "students": [
390 {'user':'ramsayb2',
391 'displayMsg': 'Brian Ramsay',
392 'errorMsg': ''},
393 {'user':'B0073235',
394 'displayMsg': '',
395 'errorMsg': 'ERROR: B0073235 does not exist!'}]
396 },
397 "CSC 226": {
398 "displayMsg": "CSC 226 matched to existing course 'Data Structures'.",
399 "errorMsg": "",
400 "students": [
401 {'user':'ramsayb2',
402 'displayMsg': 'Brian Ramsay',
403 'errorMsg': ''},
404 {'user':'lamichhanes',
405 'displayMsg': 'Sandesh Lamichhane',
406 'errorMsg': ''}]
407 }
408 }
409 }
410 }
411 """
412 excelData = load_workbook(filename=filePath)
413 excelSheet = excelData.active
415 result = {}
416 errors = []
417 term = ''
418 course = ''
419 cellRow = 0
421 for row in excelSheet.iter_rows():
422 cellRow += 1
423 cellVal = row[0].value
424 if not cellVal:
425 continue
427 # Look for a Term. Examples: Fall 2020 or Spring B 2021
428 if regex.search(r"\b[a-zA-Z]{3,}( [AB])? \d{4}\b", str(cellVal)):
429 errorMsg = ''
430 if "Spring A" in cellVal or "Spring B" in cellVal:
431 cellVal = "Spring " + cellVal.split()[-1]
432 if "Fall A" in cellVal or "Fall B" in cellVal:
433 cellVal = "Fall " + cellVal.split()[-1]
435 if cellVal.split()[0] not in ["Summer", "Spring", "Fall", "May"]:
436 errorMsg = f"ERROR: '{cellVal}' is not a valid term in row {cellRow}."
437 else:
438 latestTerm = Term.select().order_by(Term.termOrder.desc()).get()
439 isFutureTerm = latestTerm.termOrder < Term.convertDescriptionToTermOrder(cellVal)
440 if isFutureTerm:
441 errors.append(f"ERROR: '{cellVal}' is a future term in row {cellRow}.")
442 else:
443 validTerm = Term.get_or_none(Term.description == cellVal)
445 term = cellVal
446 result[term] = {
447 'displayMsg': term,
448 'errorMsg': errorMsg,
449 'courses': {}
450 }
451 if errorMsg:
452 errors.append((errorMsg,0))
454 # Look for a Course. Examples: FRN134 CSC 226
455 elif regex.search(r"\b[A-Z]{2,4} ?\d{3}\b", str(cellVal)):
456 errorMsg = displayMsg = ''
457 if not term:
458 displayMsg = cellVal
459 errorMsg = "ERROR: No term was given for this course"
460 else:
461 existingCourse = Course.get_or_none(Course.courseAbbreviation == cellVal)
462 displayMsg = f'{cellVal} will be created.'
463 if existingCourse:
464 displayMsg = f"{cellVal} matched to the existing course {existingCourse.courseName}."
466 course = cellVal
467 result[term]['courses'][course] = {
468 'displayMsg': displayMsg,
469 'errorMsg': errorMsg,
470 'students': []
471 }
472 if errorMsg:
473 errors.append((errorMsg,0))
475 # Look for a B-Number. Example: B00123456
476 elif regex.search(r"\b[B]\d{8}\b", str(cellVal)):
477 errorMsg = displayMsg = ''
478 if not course:
479 errorMsg = "ERROR: No course is connected to this student"
480 else:
481 existingUser = User.get_or_none(User.bnumber == cellVal)
482 if existingUser:
483 displayMsg = f"{existingUser.firstName} {existingUser.lastName}"
484 existingUser = existingUser.username
485 else:
486 errorMsg = f'ERROR: {row[1].value} with B# "{row[0].value}" does not exist.'
488 result[term]['courses'][course]['students'].append({
489 'user': (existingUser or cellVal),
490 'displayMsg': displayMsg,
491 'errorMsg': errorMsg})
492 if errorMsg:
493 errors.append((errorMsg,0))
495 elif cellVal: # but didn't match the regex
496 errors.append((f'ERROR: "{cellVal}" in row {cellRow} of the Excel document does not appear to be a term, course, or valid B#.',1))
499 return result, errors