Coverage for app/logic/serviceLearningCourses.py: 84%

194 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-06-21 18:28 +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 

7 

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 

23 

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)) 

37 

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] 

42 

43 

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 

51 

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 

58 

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 

63 

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] 

75 

76 for userDict in courseInfo['students']: 

77 if userDict['errorMsg']: 

78 print(f"Unable to save student. {userDict['errorMsg']}") 

79 continue 

80 

81 CourseParticipant.get_or_create(user=userDict['user'], 

82 course=courseObj, 

83 hoursEarned=20) 

84 

85def unapprovedCourses(termId: int) -> List[Course]: 

86 """ 

87 Queries the database to get all the neccessary information for 

88 submitted/unapproved courses. 

89 """ 

90 

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)) 

100 

101 return unapprovedCourses 

102 

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)) 

115 

116 return approvedCourses 

117 

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)) 

130 

131 return importedCourses 

132 

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) 

142 

143 for instructor in instructors: 

144 if instructor.course.courseName not in instructorToCoursesMap[instructor.user]: 

145 instructorToCoursesMap[instructor.user].append(instructor.course.courseName) 

146 

147 return dict(instructorToCoursesMap) 

148 

149########### Course Actions ########### 

150 

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) 

168 

169 instructors = CourseInstructor.select().where(CourseInstructor.course==oldCourse.id) 

170 for instructor in instructors: 

171 CourseInstructor.create(course=newCourse.id, 

172 user=instructor.user) 

173 

174 return newCourse 

175 

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 """ 

182 

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) 

188 

189 except DoesNotExist: 

190 print(f"File, {AttachmentUpload.fileName}, does not exist.") 

191 

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() 

203 

204 createActivityLog(f"Withdrew SLC proposal: {courseName}") 

205 

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) 

213 

214 return course 

215 

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']) 

224 

225 for toggler in ["slSectionsToggle", "permanentDesignation"]: 

226 courseData.setdefault(toggler, "off") 

227 

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()) 

242 

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()) 

248 

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) 

254 

255 # save attachments to course if applicable 

256 if attachments: 

257 addFile: FileHandler = FileHandler(attachments, courseId=course.id) 

258 addFile.saveFiles() 

259 

260 createActivityLog(f"Saved SLC proposal: {courseData['courseName']}") 

261 

262 return Course.get_by_id(course.id) 

263 

264 except Exception as e: 

265 print(e) 

266 transaction.rollback() 

267 return False 

268 

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 """ 

275 

276 with mainDB.atomic() as transaction: 

277 try: 

278 course = Course.get_by_id(courseData["courseId"]) 

279 

280 Course.update(courseName=courseData["courseName"]).where(Course.id == course.id).execute() 

281 

282 (CourseParticipant.update(hoursEarned=courseData["hoursEarned"]) 

283 .where(CourseParticipant.course_id == course.id).execute()) 

284 

285 instructorList = [] 

286 CourseInstructor.delete().where(CourseInstructor.course == course).execute() 

287 

288 if 'instructor[]' in courseData: 

289 instructorList = courseData.getlist('instructor[]') 

290 

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) 

295 

296 return Course.get_by_id(course.id) 

297 

298 except Exception as e: 

299 print(e) 

300 transaction.rollback() 

301 return False 

302 

303########### Course Actions ########### 

304 

305def parseUploadedFile(filePath): 

306 """ 

307 Parse an Excel document at the given `filePath` for courses and 

308 course participants. 

309 

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 

355 

356 result = {} 

357 errors = [] 

358 term = '' 

359 course = '' 

360 cellRow = 0 

361 

362 for row in excelSheet.iter_rows(): 

363 cellRow += 1 

364 cellVal = row[0].value 

365 if not cellVal: 

366 continue 

367 

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] 

375 

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) 

385 

386 term = cellVal 

387 result[term] = { 

388 'displayMsg': term, 

389 'errorMsg': errorMsg, 

390 'courses': {} 

391 } 

392 if errorMsg: 

393 errors.append((errorMsg,0)) 

394 

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}." 

406 

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)) 

415 

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.' 

428 

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)) 

435 

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)) 

438 

439 

440 return result, errors