Coverage for course_modules/models.py: 81%

1""" 

2Course module model. 

3""" 

4 

5from datetime import datetime, timedelta 

6from html import escape 

7import tempfile 

8import uuid 

9import pandas as pd 

10from flask import send_file, jsonify 

11 

12from core import handlers 

13 

14# Cache to store modules and the last update time 

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

16 

17 

18class Module: 

19 """Module data model""" 

20 

21 def add_module(self, module): 

22 """Adds a module to the database.""" 

23 from app import DATABASE_MANAGER 

24 

25 # module = { 

26 # "_id": uuid.uuid1().hex, 

27 # "module_id": request.form.get("module_id"), 

28 # "module_name": request.form.get("module_name"), 

29 # "module_description": request.form.get("module_description"), 

30 # } 

31 

32 if DATABASE_MANAGER.get_one_by_field( 

33 "modules", "module_id", module["module_id"] 

34 ): 

35 return jsonify({"error": "module already in database"}), 400 

36 

37 DATABASE_MANAGER.insert("modules", module) 

38 

39 if module: 

40 # Update cache 

41 modules = DATABASE_MANAGER.get_all("modules") 

42 modules_cache["data"] = modules 

43 modules_cache["last_updated"] = datetime.now() 

44 return jsonify(module), 200 

45 

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

47 

48 def delete_module_by_id(self, module_id): 

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

50 from app import DATABASE_MANAGER 

51 

52 module = DATABASE_MANAGER.get_one_by_field("modules", "module_id", module_id) 

53 

54 if not module: 

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

56 

57 DATABASE_MANAGER.delete_by_id("modules", module["_id"]) 

58 

59 students = DATABASE_MANAGER.get_all("students") 

60 for student in students: 

61 if "modules" in student and module_id in student["modules"]: 

62 student["modules"].remove(module_id) 

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

64 

65 opportunities = DATABASE_MANAGER.get_all("opportunities") 

66 for opportunity in opportunities: 

67 if ( 

68 "modules_required" in opportunity 

69 and module_id in opportunity["modules_required"] 

70 ): 

71 opportunity["modules_required"].remove(module_id) 

72 DATABASE_MANAGER.update_one_by_id( 

73 "opportunities", opportunity["_id"], opportunity 

74 ) 

75 

76 # Update cache 

77 modules = DATABASE_MANAGER.get_all("modules") 

78 modules_cache["data"] = modules 

79 modules_cache["last_updated"] = datetime.now() 

80 

81 return jsonify(module), 200 

82 

83 def delete_module_by_uuid(self, uuid): 

84 """Deletes a module from the database.""" 

85 

86 from app import DATABASE_MANAGER 

87 

88 module = DATABASE_MANAGER.get_one_by_id("modules", uuid) 

89 

90 if not module: 

91 return jsonify({"error": "module not found"}), 404 

92 

93 DATABASE_MANAGER.delete_by_id("modules", uuid) 

94 

95 students = DATABASE_MANAGER.get_all("students") 

96 for student in students: 

97 if "modules" in student and module["module_id"] in student["modules"]: 

98 student["modules"].remove(module["module_id"]) 

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

100 

101 opportunities = DATABASE_MANAGER.get_all("opportunities") 

102 for opportunity in opportunities: 

103 if ( 

104 "modules_required" in opportunity 

105 and uuid in opportunity["modules_required"] 

106 ): 

107 opportunity["modules_required"].remove(module["module_id"]) 

108 DATABASE_MANAGER.update_one_by_id( 

109 "opportunities", opportunity["_id"], opportunity 

110 ) 

111 

112 # Update cache 

113 modules = DATABASE_MANAGER.get_all("modules") 

114 modules_cache["data"] = modules 

115 modules_cache["last_updated"] = datetime.now() 

116 

117 return jsonify(module), 200 

118 

119 def get_module_by_id(self, module_id): 

120 """Retrieves a module by its ID.""" 

121 from app import DATABASE_MANAGER 

122 

123 module = DATABASE_MANAGER.get_one_by_field("modules", "module_id", module_id) 

124 

125 if module: 

126 return module 

127 

128 return None 

129 

130 def get_module_by_uuid(self, uuid): 

131 """Retrieves a module by its ID.""" 

132 from app import DATABASE_MANAGER 

133 

134 module = DATABASE_MANAGER.get_one_by_id("modules", uuid) 

135 

136 if module: 

137 return module 

138 

139 return None 

140 

141 def get_module_name_by_id(self, module_id): 

142 """Get module name by id""" 

143 module = self.get_module_by_id(module_id) 

144 if not module: 

145 return None 

146 return module["module_name"] 

147 

148 def get_modules(self): 

149 """Retrieves all modules.""" 

150 current_time = datetime.now() 

151 one_week_ago = current_time - timedelta(weeks=1) 

152 from app import DATABASE_MANAGER 

153 

154 # Check if cache is valid 

155 if ( 

156 modules_cache["data"] 

157 and modules_cache["last_updated"] 

158 and modules_cache["last_updated"] > one_week_ago 

159 ): 

160 return modules_cache["data"] 

161 

162 # Fetch modules from the database 

163 modules = DATABASE_MANAGER.get_all("modules") 

164 

165 if modules: 

166 # Update cache 

167 modules_cache["data"] = modules 

168 modules_cache["last_updated"] = current_time 

169 return modules 

170 

171 return [] 

172 

173 def get_modules_map(self): 

174 """Get modules map""" 

175 modules = self.get_modules() 

176 return {module["module_id"]: module for module in modules} 

177 

178 def reset_cache(self): 

179 """Reset cache""" 

180 from app import DATABASE_MANAGER 

181 

182 modules_cache["data"] = DATABASE_MANAGER.get_all("modules") 

183 modules_cache["last_updated"] = datetime.now() 

184 

185 def update_module_by_uuid(self, uuid, module_id, module_name, module_description): 

186 """Updates a module in the database.""" 

187 

188 from app import DATABASE_MANAGER 

189 

190 original_module = DATABASE_MANAGER.get_one_by_id("modules", uuid) 

191 if not DATABASE_MANAGER.get_one_by_id("modules", uuid): 

192 return jsonify({"error": "module not found"}), 404 

193 

194 updated_module = { 

195 "_id": uuid, 

196 "module_id": module_id, 

197 "module_name": module_name, 

198 "module_description": module_description, 

199 } 

200 

201 DATABASE_MANAGER.update_one_by_id("modules", uuid, updated_module) 

202 

203 students = DATABASE_MANAGER.get_all("students") 

204 for student in students: 

205 if ( 

206 "modules" in student 

207 and original_module["module_id"] in student["modules"] 

208 ): 

209 student["modules"].remove(original_module["module_id"]) 

210 student["modules"].append(module_id) 

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

212 

213 opportunities = DATABASE_MANAGER.get_all("opportunities") 

214 for opportunity in opportunities: 

215 if ( 

216 "modules_required" in opportunity 

217 and original_module["module_id"] in opportunity["modules_required"] 

218 ): 

219 opportunity["modules_required"].remove(original_module["module_id"]) 

220 opportunity["modules_required"].append(module_id) 

221 DATABASE_MANAGER.update_one_by_id( 

222 "opportunities", opportunity["_id"], opportunity 

223 ) 

224 

225 # Update cache 

226 modules = DATABASE_MANAGER.get_all("modules") 

227 modules_cache["data"] = modules 

228 modules_cache["last_updated"] = datetime.now() 

229 

230 return jsonify({"message": "Updated"}), 200 

231 

232 def delete_all_modules(self): 

233 """Deletes all modules from the database.""" 

234 from app import DATABASE_MANAGER 

235 

236 DATABASE_MANAGER.delete_all("modules") 

237 modules_cache["data"] = [] 

238 modules_cache["last_updated"] = datetime.now() 

239 

240 students = DATABASE_MANAGER.get_all("students") 

241 DATABASE_MANAGER.delete_all("students") 

242 updated_students = [] 

243 for student in students: 

244 if "modules" in student: 

245 student["modules"] = [] 

246 updated_students.append(student) 

247 

248 if updated_students: 

249 DATABASE_MANAGER.insert_many("students", updated_students) 

250 

251 opportunities = DATABASE_MANAGER.get_all("opportunities") 

252 DATABASE_MANAGER.delete_all("opportunities") 

253 updated_opportunities = [] 

254 for opp in opportunities: 

255 opp["modules_required"] = [] 

256 updated_opportunities.append(opp) 

257 if updated_opportunities: 

258 DATABASE_MANAGER.insert_many("opportunities", updated_opportunities) 

259 

260 return jsonify({"message": "Deleted"}), 200 

261 

262 def download_all_modules(self): 

263 """Download all modules""" 

264 from app import DATABASE_MANAGER 

265 

266 modules = DATABASE_MANAGER.get_all("modules") 

267 # Create a DataFrame from the modules 

268 df = pd.DataFrame(modules) 

269 

270 # Use tempfile to create a temporary file 

271 df.drop(columns=["_id"], inplace=True) 

272 with tempfile.NamedTemporaryFile(suffix=".xlsx") as tmp: 

273 file_path = tmp.name 

274 # Save the DataFrame to an Excel file 

275 df.to_excel( 

276 file_path, 

277 index=False, 

278 header=["Module_id", "Module_name", "Module_description"], 

279 ) 

280 

281 return send_file( 

282 file_path, download_name="modules.xlsx", as_attachment=True 

283 ) 

284 

285 def upload_course_modules(self, file): 

286 """Add course modules from an Excel file.""" 

287 

288 from app import DATABASE_MANAGER 

289 

290 # Read the Excel file 

291 try: 

292 df = handlers.excel_verifier_and_reader( 

293 file, 

294 {"Module_id", "Module_name", "Module_description"}, 

295 ) 

296 except Exception as e: 

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

298 

299 # Convert the DataFrame to a list of dictionaries 

300 modules = df.to_dict(orient="records") 

301 

302 clean_data = [] 

303 current_ids = set( 

304 module["module_id"] for module in DATABASE_MANAGER.get_all("modules") 

305 ) 

306 

307 ids = set() 

308 

309 for i, module in enumerate(modules): 

310 temp = { 

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

312 "module_id": escape(module.get("Module_id", "")), 

313 "module_name": escape(module.get("Module_name", "")), 

314 "module_description": escape(module.get("Module_description", "")), 

315 } 

316 if not temp["module_id"] or not temp["module_name"]: 

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

318 if temp["module_id"] in ids: 

319 return ( 

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

321 400, 

322 ) 

323 if temp["module_id"] in current_ids: 

324 return jsonify({"error": "Module already in database"}), 400 

325 clean_data.append(temp) 

326 ids.add(temp["module_id"]) 

327 

328 DATABASE_MANAGER.insert_many("modules", clean_data) 

329 

330 # Update cache 

331 modules = DATABASE_MANAGER.get_all("modules") 

332 modules_cache["data"] = modules 

333 modules_cache["last_updated"] = datetime.now() 

334 

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