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