Coverage for app/logic/serviceLearningCourses.py: 84%
203 statements
« prev ^ index » next coverage.py v7.2.7, created at 2024-07-24 12:19 +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 termObj: Term = Term.get_or_none(description = term) or addPastTerm(term)
71 if not termObj:
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 courseObj: Course = Course.get_or_create(
81 courseAbbreviation = course,
82 term = termObj,
83 defaults = {"CourseName" : "",
84 "sectionDesignation" : "",
85 "courseCredit" : "1",
86 "term" : termObj,
87 "status" : 4,
88 "createdBy" : g.current_user,
89 "serviceLearningDesignatedSections" : "",
90 "previouslyApprovedDescription" : "" })[0]
92 for userDict in courseInfo['students']:
93 if userDict['errorMsg']:
94 print(f"Unable to save student. {userDict['errorMsg']}")
95 continue
97 CourseParticipant.get_or_create(user=userDict['user'],
98 course=courseObj,
99 hoursEarned=20)
101def unapprovedCourses(termId: int) -> List[Course]:
102 """
103 Queries the database to get all the neccessary information for
104 submitted/unapproved courses.
105 """
107 unapprovedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
108 .join(CourseInstructor, JOIN.LEFT_OUTER)
109 .join(User, JOIN.LEFT_OUTER).switch(Course)
110 .join(CourseStatus).switch(Course)
111 .join(Term)
112 .where(Term.id == termId,
113 Course.status.in_([CourseStatus.SUBMITTED, CourseStatus.IN_PROGRESS]))
114 .group_by(Course, Term, CourseStatus)
115 .order_by(Course.status))
117 return unapprovedCourses
119def approvedCourses(termId: int) -> List[Course]:
120 """
121 Queries the database to get all the necessary information for
122 approved courses.
123 """
124 approvedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
125 .join(CourseInstructor, JOIN.LEFT_OUTER)
126 .join(User, JOIN.LEFT_OUTER).switch(Course)
127 .join(CourseStatus).switch(Course)
128 .join(Term)
129 .where(Term.id == termId, Course.status == CourseStatus.APPROVED)
130 .group_by(Course, Term, CourseStatus))
132 return approvedCourses
134def getImportedCourses(termId: int) -> List[Course]:
135 """
136 Queries the database to get all the necessary information for
137 imported courses.
138 """
139 importedCourses: List[Course] = list(Course.select(Course, Term, CourseStatus, fn.GROUP_CONCAT(" " ,User.firstName, " ", User.lastName).alias('instructors'))
140 .join(CourseInstructor, JOIN.LEFT_OUTER)
141 .join(User, JOIN.LEFT_OUTER).switch(Course)
142 .join(CourseStatus).switch(Course)
143 .join(Term)
144 .where(Term.id == termId, Course.status == CourseStatus.IMPORTED)
145 .group_by(Course, Term, CourseStatus))
147 return importedCourses
149def getInstructorCourses() -> Dict[User, str]:
150 """
151 This function queries all of the course instructors and their classes and maps
152 each instructor to its respective courses.
153 """
154 instructors: List[CourseInstructor] = list(CourseInstructor.select(CourseInstructor, User, Course)
155 .join(User).switch()
156 .join(Course))
157 instructorToCoursesMap: DefaultDict[User, str] = defaultdict(list)
159 for instructor in instructors:
160 if instructor.course.courseName not in instructorToCoursesMap[instructor.user]:
161 instructorToCoursesMap[instructor.user].append(instructor.course.courseName)
163 return dict(instructorToCoursesMap)
165########### Course Actions ###########
167def renewProposal(courseID, term) -> Course:
168 """
169 Renews proposal of ID passed in for the selected term.
170 Sets status to in progress.
171 """
172 oldCourse: Course = Course.get_by_id(courseID)
173 newCourse: Course = Course.get_by_id(courseID)
174 newCourse.id = None
175 newCourse.term = Term.get_by_id(term)
176 newCourse.status = CourseStatus.IN_PROGRESS
177 newCourse.isPreviouslyApproved = True
178 newCourse.save()
179 questions: List[CourseQuestion] = list(CourseQuestion.select().where(CourseQuestion.course==oldCourse))
180 for question in questions:
181 CourseQuestion.create(course=newCourse.id,
182 questionContent=question.questionContent,
183 questionNumber=question.questionNumber)
185 instructors = CourseInstructor.select().where(CourseInstructor.course==oldCourse.id)
186 for instructor in instructors:
187 CourseInstructor.create(course=newCourse.id,
188 user=instructor.user)
190 return newCourse
192def withdrawProposal(courseID) -> None:
193 """
194 Withdraws proposal of ID passed in. Removes foreign keys first.
195 Key Dependencies: QuestionNote, CourseQuestion, CourseParticipant,
196 CourseInstructor, Note
197 """
199 # delete syllabus
200 try:
201 syllabi: List[AttachmentUpload] = list(AttachmentUpload.select().where(AttachmentUpload.course==courseID))
202 for syllabus in syllabi:
203 FileHandler(courseId = courseID).deleteFile(syllabus.id)
205 except DoesNotExist:
206 print(f"File, {AttachmentUpload.fileName}, does not exist.")
208 # delete course object
209 course: Course = Course.get(Course.id == courseID)
210 courseName: str = course.courseName
211 questions: List[CourseQuestion] = CourseQuestion.select().where(CourseQuestion.course == course)
212 notes: List[Note] = list(Note.select(Note.id)
213 .join(QuestionNote)
214 .where(QuestionNote.question.in_(questions))
215 .distinct())
216 course.delete_instance(recursive=True)
217 for note in notes:
218 note.delete_instance()
220 createActivityLog(f"Withdrew SLC proposal: {courseName}")
222def createCourse(creator: str="No user provided") -> Course:
223 """
224 Creates and returns an empty, in progress course.
225 """
226 course: Course = Course.create(status=CourseStatus.IN_PROGRESS, createdBy=creator)
227 for number in range(1, 7):
228 CourseQuestion.create(course=course, questionNumber=number)
230 return course
232def updateCourse(courseData, attachments=None) -> Union[Course, bool]:
233 """
234 This function will take in courseData for the SLC proposal page and a dictionary
235 of instuctors assigned to the course and update the information in the db.
236 """
237 with mainDB.atomic() as transaction:
238 try:
239 course: Course = Course.get_by_id(courseData['courseID'])
241 for toggler in ["slSectionsToggle", "permanentDesignation"]:
242 courseData.setdefault(toggler, "off")
244 (Course.update(courseName=courseData["courseName"],
245 courseAbbreviation=courseData["courseAbbreviation"],
246 sectionDesignation=courseData["sectionDesignation"],
247 courseCredit=courseData["credit"],
248 isRegularlyOccurring=int(courseData["isRegularlyOccurring"]),
249 term=courseData['term'],
250 status=CourseStatus.SUBMITTED,
251 isPreviouslyApproved=int(courseData["isPreviouslyApproved"]),
252 previouslyApprovedDescription = courseData["previouslyApprovedDescription"],
253 isAllSectionsServiceLearning=("on" in courseData["slSectionsToggle"]),
254 serviceLearningDesignatedSections=courseData["slDesignation"],
255 isPermanentlyDesignated=("on" in courseData["permanentDesignation"]),
256 hasSlcComponent = int(courseData["hasSlcComponent"]))
257 .where(Course.id == course.id).execute())
259 # update the existing entry with the new question responses
260 for questionIndex in range(1, 7):
261 (CourseQuestion.update(questionContent=courseData[f"{questionIndex}"])
262 .where((CourseQuestion.questionNumber == questionIndex) &
263 (CourseQuestion.course==course)).execute())
265 # delete all course instructors and create entries for the updated instructors
266 CourseInstructor.delete().where(CourseInstructor.course == course).execute()
267 instructorList: List[str] = courseData.getlist('instructor[]')
268 for instructor in instructorList:
269 CourseInstructor.create(course=course, user=instructor)
271 # save attachments to course if applicable
272 if attachments:
273 addFile: FileHandler = FileHandler(attachments, courseId=course.id)
274 addFile.saveFiles()
276 createActivityLog(f"Saved SLC proposal: {courseData['courseName']}")
278 return Course.get_by_id(course.id)
280 except Exception as e:
281 print(e)
282 transaction.rollback()
283 return False
285def editImportedCourses(courseData):
286 """
287 This function will take in courseData for the SLC proposal page and a dictionary
288 of instructors assigned to the imported course after that one is edited
289 and update the information in the db.
290 """
292 with mainDB.atomic() as transaction:
293 try:
294 course = Course.get_by_id(courseData["courseId"])
296 Course.update(courseName=courseData["courseName"]).where(Course.id == course.id).execute()
298 (CourseParticipant.update(hoursEarned=courseData["hoursEarned"])
299 .where(CourseParticipant.course_id == course.id).execute())
301 instructorList = []
302 CourseInstructor.delete().where(CourseInstructor.course == course).execute()
304 if 'instructor[]' in courseData:
305 instructorList = courseData.getlist('instructor[]')
307 for instructor in instructorList:
308 # Checks that empty string is not added as a course instructor because some keys in the dictionary are empty string.
309 if instructor:
310 CourseInstructor.create(course=course, user=instructor)
312 return Course.get_by_id(course.id)
314 except Exception as e:
315 print(e)
316 transaction.rollback()
317 return False
319########### Course Actions ###########
321def parseUploadedFile(filePath):
322 """
323 Parse an Excel document at the given `filePath` for courses and
324 course participants.
326 The return value will be a tuple. The second value is a list of
327 error message tuples. The first tuple value is the error message,
328 and the second is a 0 or 1 indicating whether the error is a
329 'general' error - 1 - or an error on a specific course, term, or
330 person. The first value is a dictionary keyed by the term
331 description. Each value is another dictionary, with a key for
332 'displayMsg' and 'errorMsg', and a 'courses' key whose value is
333 a dictionary with keys for the courses in the term. Each course
334 has a 'displayMsg' and 'errorMsg' key, and a 'students' key
335 that has a list of dictionaries with 'user', 'displayMsg', and
336 'errorMsg' keys.
337 E.g.,
338 {
339 "Fall 2021": {
340 "displayMsg": "",
341 "errorMsg": "",
342 "courses": {
343 "CSC 330": {
344 "displayMsg: "CSC 330 will be created",
345 "errorMsg: "",
346 "students": [
347 {'user':'ramsayb2',
348 'displayMsg': 'Brian Ramsay',
349 'errorMsg': ''},
350 {'user':'B0073235',
351 'displayMsg': '',
352 'errorMsg': 'ERROR: B0073235 does not exist!'}]
353 },
354 "CSC 226": {
355 "displayMsg": "CSC 226 matched to existing course 'Data Structures'.",
356 "errorMsg": "",
357 "students": [
358 {'user':'ramsayb2',
359 'displayMsg': 'Brian Ramsay',
360 'errorMsg': ''},
361 {'user':'lamichhanes',
362 'displayMsg': 'Sandesh Lamichhane',
363 'errorMsg': ''}]
364 }
365 }
366 }
367 }
368 """
369 excelData = load_workbook(filename=filePath)
370 excelSheet = excelData.active
372 result = {}
373 errors = []
374 term = ''
375 course = ''
376 cellRow = 0
378 for row in excelSheet.iter_rows():
379 cellRow += 1
380 cellVal = row[0].value
381 if not cellVal:
382 continue
384 # Look for a Term. Examples: Fall 2020 or Spring B 2021
385 if regex.search(r"\b[a-zA-Z]{3,}( [AB])? \d{4}\b", str(cellVal)):
386 errorMsg = ''
387 if "Spring A" in cellVal or "Spring B" in cellVal:
388 cellVal = "Spring " + cellVal.split()[-1]
389 if "Fall A" in cellVal or "Fall B" in cellVal:
390 cellVal = "Fall " + cellVal.split()[-1]
392 if cellVal.split()[0] not in ["Summer", "Spring", "Fall", "May"]:
393 errorMsg = f"ERROR: '{cellVal}' is not a valid term in row {cellRow}."
394 else:
395 latestTerm = Term.select().order_by(Term.termOrder.desc()).get()
396 isFutureTerm = latestTerm.termOrder < Term.convertDescriptionToTermOrder(cellVal)
397 if isFutureTerm:
398 errors.append(f"ERROR: '{cellVal}' is a future term in row {cellRow}.")
399 else:
400 validTerm = Term.get_or_none(Term.description == cellVal)
402 term = cellVal
403 result[term] = {
404 'displayMsg': term,
405 'errorMsg': errorMsg,
406 'courses': {}
407 }
408 if errorMsg:
409 errors.append((errorMsg,0))
411 # Look for a Course. Examples: FRN134 CSC 226
412 elif regex.search(r"\b[A-Z]{2,4} ?\d{3}\b", str(cellVal)):
413 errorMsg = displayMsg = ''
414 if not term:
415 displayMsg = cellVal
416 errorMsg = "ERROR: No term was given for this course"
417 else:
418 existingCourse = Course.get_or_none(Course.courseAbbreviation == cellVal)
419 displayMsg = f'{cellVal} will be created.'
420 if existingCourse:
421 displayMsg = f"{cellVal} matched to the existing course {existingCourse.courseName}."
423 course = cellVal
424 result[term]['courses'][course] = {
425 'displayMsg': displayMsg,
426 'errorMsg': errorMsg,
427 'students': []
428 }
429 if errorMsg:
430 errors.append((errorMsg,0))
432 # Look for a B-Number. Example: B00123456
433 elif regex.search(r"\b[B]\d{8}\b", str(cellVal)):
434 errorMsg = displayMsg = ''
435 if not course:
436 errorMsg = "ERROR: No course is connected to this student"
437 else:
438 existingUser = User.get_or_none(User.bnumber == cellVal)
439 if existingUser:
440 displayMsg = f"{existingUser.firstName} {existingUser.lastName}"
441 existingUser = existingUser.username
442 else:
443 errorMsg = f'ERROR: {row[1].value} with B# "{row[0].value}" does not exist.'
445 result[term]['courses'][course]['students'].append({
446 'user': (existingUser or cellVal),
447 'displayMsg': displayMsg,
448 'errorMsg': errorMsg})
449 if errorMsg:
450 errors.append((errorMsg,0))
452 elif cellVal: # but didn't match the regex
453 errors.append((f'ERROR: "{cellVal}" in row {cellRow} of the Excel document does not appear to be a term, course, or valid B#.',1))
456 return result, errors