Coverage for courses/models.py: 88%

1""" 

2Courses model.""" 

3 

4from datetime import datetime, timedelta 

5from html import escape 

6import tempfile 

7import uuid 

8from flask import jsonify, send_file 

9import pandas as pd 

10 

11from core import handlers 

12 

13 

14# Cache to store courses and the last update time 

15courses_cache = {"data": None, "last_updated": None} 

16 

17 

18class Course: 

19 """Course data model""" 

20 

21 def reset_cache(self): 

22 """Resets the courses cache.""" 

23 from app import DATABASE_MANAGER 

24 

25 courses = DATABASE_MANAGER.get_all("courses") 

26 courses_cache["data"] = courses 

27 courses_cache["last_updated"] = datetime.now() 

28 return courses 

29 

30 def add_course(self, course): 

31 """Adds a course to the database.""" 

32 from app import DATABASE_MANAGER 

33 

34 if DATABASE_MANAGER.get_one_by_field( 

35 "courses", "course_id", course["course_id"] 

36 ): 

37 return jsonify({"error": "Course already in database"}), 400 

38 

39 DATABASE_MANAGER.insert("courses", course) 

40 

41 if course: 

42 # Update cache 

43 self.reset_cache() 

44 return jsonify(course), 200 

45 

46 return jsonify({"error": "Course not added"}), 400 

47 

48 def delete_course_by_uuid(self, id_val): 

49 """Deletes a course from the database.""" 

50 from app import DATABASE_MANAGER 

51 

52 course = DATABASE_MANAGER.get_one_by_id("courses", id_val) 

53 

54 if not course: 

55 return jsonify({"error": "Course not found"}), 404 

56 

57 students = DATABASE_MANAGER.get_all_by_field( 

58 "students", "course", course["course_id"] 

59 ) 

60 

61 if students and len(students) > 0: 

62 return jsonify({"error": "Course has students enrolled"}), 400 

63 

64 opportunities = DATABASE_MANAGER.get_all_by_in_list( 

65 "opportunities", "courses_required", [course["course_id"]] 

66 ) 

67 

68 for opportunity in opportunities: 

69 if ( 

70 "courses_required" in opportunity 

71 and course["course_id"] in opportunity["courses_required"] 

72 ): 

73 opportunity["courses_required"].remove(course["course_id"]) 

74 DATABASE_MANAGER.update_one_by_id( 

75 "opportunities", opportunity["_id"], opportunity 

76 ) 

77 

78 DATABASE_MANAGER.delete_by_id("courses", course["_id"]) 

79 # Update cache 

80 self.reset_cache() 

81 

82 return jsonify(course), 200 

83 

84 def get_course_by_id(self, course_id): 

85 """Retrieves a course by its ID.""" 

86 from app import DATABASE_MANAGER 

87 

88 course = DATABASE_MANAGER.get_one_by_field("courses", "course_id", course_id) 

89 

90 if course: 

91 return course 

92 

93 return None 

94 

95 def get_course_by_uuid(self, uuid): 

96 """Get course by uuid""" 

97 from app import DATABASE_MANAGER 

98 

99 course = DATABASE_MANAGER.get_one_by_id("courses", uuid) 

100 if course: 

101 return course 

102 

103 return None 

104 

105 def get_course_name_by_id(self, course_id): 

106 """Get course name by id""" 

107 course = self.get_course_by_id(course_id) 

108 if not course: 

109 return None 

110 return course["course_name"] 

111 

112 def get_courses(self): 

113 """Retrieves all courses.""" 

114 from app import DATABASE_MANAGER 

115 

116 current_time = datetime.now() 

117 one_week_ago = current_time - timedelta(weeks=1) 

118 

119 # Check if cache is valid 

120 if ( 

121 courses_cache["data"] 

122 and courses_cache["last_updated"] 

123 and courses_cache["last_updated"] > one_week_ago 

124 ): 

125 return courses_cache["data"] 

126 

127 # Fetch courses from the database 

128 courses = DATABASE_MANAGER.get_all("courses") 

129 

130 if courses: 

131 # Update cache 

132 self.reset_cache() 

133 return courses 

134 

135 return [] 

136 

137 def get_courses_map(self): 

138 """Get courses map""" 

139 courses = self.get_courses() 

140 return {course["course_id"]: course for course in courses} 

141 

142 def update_course(self, uuid, updated_course): 

143 """Update course""" 

144 from app import DATABASE_MANAGER 

145 

146 original = DATABASE_MANAGER.get_one_by_id("courses", uuid) 

147 if not original: 

148 return jsonify({"error": "Course not found"}), 404 

149 

150 if ( 

151 "course_id" in updated_course 

152 and updated_course["course_id"] != original["course_id"] 

153 ): 

154 match = DATABASE_MANAGER.get_one_by_field( 

155 "courses", "course_id", updated_course["course_id"] 

156 ) 

157 if match: 

158 return jsonify({"error": "Course ID already exists"}), 400 

159 DATABASE_MANAGER.update_one_by_id("courses", uuid, updated_course) 

160 

161 students = DATABASE_MANAGER.get_all("students") 

162 for student in students: 

163 if "course" in student and original["course_id"] == student["course"]: 

164 student["course"] = updated_course["course_id"] 

165 DATABASE_MANAGER.update_one_by_id("students", student["_id"], student) 

166 

167 opportunities = DATABASE_MANAGER.get_all("opportunities") 

168 for opportunity in opportunities: 

169 if ( 

170 "courses_required" in opportunity 

171 and original["course_id"] in opportunity["courses_required"] 

172 ): 

173 opportunity["courses_required"].remove(original["course_id"]) 

174 opportunity["courses_required"].append(updated_course["course_id"]) 

175 DATABASE_MANAGER.update_one_by_id( 

176 "opportunities", opportunity["_id"], opportunity 

177 ) 

178 self.reset_cache() 

179 return jsonify({"message": "Course was updated"}), 200 

180 

181 def delete_all_courses(self): 

182 """Deletes all courses from the database.""" 

183 from app import DATABASE_MANAGER 

184 

185 DATABASE_MANAGER.delete_all("courses") 

186 courses_cache["data"] = [] 

187 courses_cache["last_updated"] = datetime.now() 

188 

189 DATABASE_MANAGER.delete_all("students") 

190 

191 opportunities = DATABASE_MANAGER.get_all("opportunities") 

192 DATABASE_MANAGER.delete_all("opportunities") 

193 updated_opportunities = [] 

194 for opp in opportunities: 

195 if "courses_required" in opp: 

196 opp["courses_required"] = [] 

197 updated_opportunities.append(opp) 

198 

199 if updated_opportunities: 

200 DATABASE_MANAGER.insert_many("opportunities", updated_opportunities) 

201 

202 return jsonify({"message": "All courses deleted"}), 200 

203 

204 def download_all_courses(self): 

205 """Download all courses""" 

206 from app import DATABASE_MANAGER 

207 

208 courses = DATABASE_MANAGER.get_all("courses") 

209 

210 for course in courses: 

211 course_data = course["course_name"].rsplit(", ", 1) 

212 course["Course_name"] = course_data[0] 

213 course["Qualification"] = course_data[1] if len(course_data) > 1 else "" 

214 course["UCAS_code"] = course.pop("course_id") 

215 course["Course_description"] = course.pop("course_description") 

216 

217 del course["_id"] 

218 # Create a DataFrame from the courses 

219 df = pd.DataFrame(courses) 

220 

221 with tempfile.NamedTemporaryFile(suffix=".xlsx") as temp_file: 

222 file_path = temp_file.name 

223 

224 # Save the DataFrame to the temporary Excel file 

225 df.to_excel( 

226 file_path, 

227 index=False, 

228 columns=[ 

229 "UCAS_code", 

230 "Course_name", 

231 "Qualification", 

232 "Course_description", 

233 ], 

234 ) 

235 

236 # Send the file as an attachment 

237 return send_file( 

238 file_path, download_name="courses.xlsx", as_attachment=True 

239 ) 

240 

241 def upload_course_data(self, file): 

242 """Add courses from an Excel file.""" 

243 from app import DATABASE_MANAGER 

244 

245 # Read the Excel file 

246 try: 

247 df = handlers.excel_verifier_and_reader( 

248 file, 

249 {"UCAS_code", "Course_name", "Qualification", "Course_description"}, 

250 ) 

251 except Exception as e: 

252 return jsonify({"error": f"Failed to read file: {str(e)}"}), 400 

253 

254 # Convert the DataFrame to a list of dictionaries 

255 courses = df.to_dict(orient="records") 

256 

257 clean_data = [] 

258 current_ids = set( 

259 course["course_id"] for course in DATABASE_MANAGER.get_all("courses") 

260 ) 

261 ids = set() 

262 

263 for i, course in enumerate(courses): 

264 temp = { 

265 "_id": uuid.uuid4().hex, 

266 "course_id": escape(course.get("UCAS_code", "")), 

267 "course_name": escape( 

268 f"{course.get('Course_name', '')}, {course.get('Qualification', '')}" 

269 ), 

270 "course_description": escape(course.get("Course_description", "")), 

271 } 

272 if not temp["course_id"] or not temp["course_name"]: 

273 return jsonify({"error": "Invalid data in row " + str(i + 1)}), 400 

274 if temp["course_id"] in ids: 

275 return ( 

276 jsonify({"error": "Duplicate course ID in row " + str(i + 1)}), 

277 400, 

278 ) 

279 if temp["course_id"] in current_ids: 

280 return jsonify({"error": "Course ID already exists"}), 400 

281 clean_data.append(temp) 

282 ids.add(temp["course_id"]) 

283 

284 DATABASE_MANAGER.insert_many("courses", clean_data) 

285 

286 # Update cache 

287 courses = DATABASE_MANAGER.get_all("courses") 

288 courses_cache["data"] = courses 

289 courses_cache["last_updated"] = datetime.now() 

290 

291 return jsonify({"message": "Uploaded"}), 200