Coverage for skills/models.py: 83%

1""" 

2Skills model. 

3""" 

4 

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 

14class Skill: 

15 """ 

16 A class used to represent and manage skills in the database.""" 

17 

18 def find_skill(self, skill_name="", skill_id=""): 

19 """Check if a skill exists in the database.""" 

20 from app import DATABASE_MANAGER 

21 

22 if skill_name: 

23 return DATABASE_MANAGER.get_one_by_field("skills", "skill_name", skill_name) 

24 if skill_id: 

25 return DATABASE_MANAGER.get_one_by_id("skills", skill_id) 

26 

27 return None 

28 

29 def add_skill(self, skill): 

30 """Add Skill to database""" 

31 # skill = { 

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

33 # "skill_name": request.form.get("skill_name"), 

34 # "skill_description": request.form.get("skill_description"), 

35 # } 

36 

37 # Check if skill already exists# 

38 from app import DATABASE_MANAGER 

39 

40 if self.find_skill(skill["skill_name"], None): 

41 return jsonify({"error": "Skill already in database"}), 400 

42 

43 DATABASE_MANAGER.insert("skills", skill) 

44 

45 return jsonify(skill), 200 

46 

47 def delete_skill(self, skill_id): 

48 """Delete kill from database""" 

49 from app import DATABASE_MANAGER 

50 

51 result = DATABASE_MANAGER.delete_by_id("skills", skill_id) 

52 

53 if result.deleted_count == 0: 

54 return jsonify({"error": "Skill not found"}), 404 

55 

56 students = DATABASE_MANAGER.get_all("students") 

57 

58 for student in students: 

59 if skill_id in student.get("skills", []): 

60 student["skills"].remove(skill_id) 

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

62 

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

64 

65 def get_skill_by_id(self, skill_id): 

66 """Get skill by ID tag""" 

67 skill = self.find_skill(None, skill_id) 

68 

69 if skill: 

70 return skill 

71 

72 return None 

73 

74 def get_skill_name_by_id(self, skill_id): 

75 """Get skill name by id""" 

76 skill = self.get_skill_by_id(skill_id) 

77 if not skill: 

78 return None 

79 return skill["skill_name"] 

80 

81 def get_skills(self): 

82 """Get full list of skills if cached get that instead""" 

83 from app import DATABASE_MANAGER 

84 

85 # Fetch skills from the database 

86 skills = DATABASE_MANAGER.get_all("skills") 

87 

88 return skills 

89 

90 def get_skills_map(self): 

91 """Get skills map""" 

92 skills = self.get_skills() 

93 return {skill["_id"]: skill for skill in skills} 

94 

95 def attempt_add_skill(self, skill_name): 

96 """Add skill to attempted skills""" 

97 from app import DATABASE_MANAGER 

98 

99 found_skill = DATABASE_MANAGER.get_one_by_field( 

100 "attempted_skills", "skill_name", skill_name 

101 ) 

102 

103 if found_skill: 

104 DATABASE_MANAGER.increment( 

105 "attempted_skills", found_skill["_id"], "used", 1 

106 ) 

107 return jsonify(found_skill), 200 

108 

109 new_skill = { 

110 "_id": uuid.uuid1().hex, 

111 "skill_name": skill_name, 

112 "skill_description": "", 

113 "used": 1, 

114 } 

115 

116 DATABASE_MANAGER.insert("attempted_skills", new_skill) 

117 return jsonify(new_skill), 200 

118 

119 def get_list_attempted_skills(self): 

120 """Get list of attempted skills""" 

121 from app import DATABASE_MANAGER 

122 

123 attempted_skills = DATABASE_MANAGER.get_all("attempted_skills") 

124 

125 if not attempted_skills: 

126 return [] 

127 

128 return attempted_skills 

129 

130 def get_attempted_skill(self, skill_id): 

131 """Get attempted skill""" 

132 from app import DATABASE_MANAGER 

133 

134 skill = DATABASE_MANAGER.get_one_by_id("attempted_skills", skill_id) 

135 

136 if not skill: 

137 return None 

138 

139 return skill 

140 

141 def approve_skill(self, skill_id, description): 

142 """Approve skill""" 

143 from app import DATABASE_MANAGER 

144 

145 skill = DATABASE_MANAGER.get_one_by_id("attempted_skills", skill_id) 

146 

147 if not skill: 

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

149 

150 del skill["used"] 

151 if description == "": 

152 return jsonify({"error": "Description is empty"}), 400 

153 skill["skill_description"] = description 

154 skill["skill_name"] = skill["skill_name"].capitalize() 

155 DATABASE_MANAGER.insert("skills", skill) 

156 DATABASE_MANAGER.delete_by_id("attempted_skills", skill_id) 

157 

158 # Update students 

159 students = DATABASE_MANAGER.get_all("students") 

160 for student in students: 

161 if skill_id in student.get("attempted_skills", []): 

162 student["skills"].append(skill_id) 

163 student["attempted_skills"].remove(skill_id) 

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

165 

166 return jsonify({"message": "Approved"}), 200 

167 

168 def reject_skill(self, skill_id): 

169 """Reject skill""" 

170 from app import DATABASE_MANAGER 

171 

172 skill = DATABASE_MANAGER.get_one_by_id("attempted_skills", skill_id) 

173 

174 if not skill: 

175 return jsonify({"error": "Attempted skill not found"}), 404 

176 

177 DATABASE_MANAGER.delete_by_id("attempted_skills", skill_id) 

178 

179 # Update students 

180 students = DATABASE_MANAGER.get_all("students") 

181 for student in students: 

182 if skill_id in student.get("attempted_skills", []): 

183 student["attempted_skills"].remove(skill_id) 

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

185 

186 return jsonify({"message": "Rejected"}), 200 

187 

188 def update_attempt_skill(self, skill_id, skill_name, skill_description): 

189 """Update attempted skill""" 

190 from app import DATABASE_MANAGER 

191 

192 skill = DATABASE_MANAGER.get_one_by_id("attempted_skills", skill_id) 

193 

194 if not skill: 

195 return jsonify({"error": "Skill not found"}), 404 

196 

197 skill["skill_name"] = skill_name 

198 skill["skill_description"] = skill_description 

199 DATABASE_MANAGER.update_one_by_id("attempted_skills", skill_id, skill) 

200 

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

202 

203 def update_skill(self, skill_id, skill_name: str, skill_description): 

204 """Updates a skill""" 

205 

206 from app import DATABASE_MANAGER 

207 

208 original = self.find_skill(None, skill_id) 

209 if not original: 

210 return jsonify({"error": "Skill not found"}), 404 

211 

212 skills = DATABASE_MANAGER.get_all_by_field( 

213 "skills", "skill_name", skill_name.capitalize() 

214 ) 

215 skills.extend( 

216 DATABASE_MANAGER.get_all_by_field("skills", "skill_name", skill_name) 

217 ) 

218 for skill in skills: 

219 if skill["_id"] != original["_id"]: 

220 return jsonify({"error": "Skill name already in use"}), 400 

221 

222 updated_skill = { 

223 "_id": skill_id, 

224 "skill_name": skill_name.capitalize(), 

225 "skill_description": skill_description, 

226 } 

227 

228 DATABASE_MANAGER.update_one_by_id("skills", skill_id, updated_skill) 

229 

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

231 

232 def download_all(self): 

233 """Returns a xlsx file with all skills""" 

234 from app import DATABASE_MANAGER 

235 

236 skills = DATABASE_MANAGER.get_all("skills") 

237 clean_data = [] 

238 

239 for skill in skills: 

240 temp = {} 

241 temp["Skill_Name"] = skill.pop("skill_name") 

242 temp["Skill_Description"] = skill.pop("skill_description") 

243 clean_data.append(temp) 

244 

245 df = pd.DataFrame(clean_data) 

246 

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

248 df.to_excel(tmp.name, index=False) 

249 tmp_file = tmp.name 

250 

251 return send_file(tmp_file, as_attachment=True, download_name="skills.xlsx") 

252 

253 def download_attempted(self): 

254 """Returns a xlsx file with all attempted skills""" 

255 from app import DATABASE_MANAGER 

256 

257 skills = DATABASE_MANAGER.get_all("attempted_skills") 

258 clean_data = [] 

259 for skill in skills: 

260 temp = {} 

261 temp["Skill_Name"] = skill.pop("skill_name") 

262 if "skill_description" in skill: 

263 temp["Skill_Description"] = skill.pop("skill_description") 

264 else: 

265 temp["Skill_Description"] = "" 

266 temp["Used"] = skill.pop("used") 

267 clean_data.append(temp) 

268 

269 df = pd.DataFrame(clean_data) 

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

271 df.to_excel(tmp.name, index=False) 

272 tmp_file = tmp.name 

273 

274 return send_file( 

275 tmp_file, as_attachment=True, download_name="attempted_skills.xlsx" 

276 ) 

277 

278 def upload_skills(self, file): 

279 """Upload skills""" 

280 

281 try: 

282 df = handlers.excel_verifier_and_reader( 

283 file, {"Skill_Name", "Skill_Description"} 

284 ) 

285 except Exception as e: 

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

287 

288 skills = df.to_dict(orient="records") 

289 

290 from app import DATABASE_MANAGER 

291 

292 current_skills = set( 

293 skill["skill_name"].lower() for skill in DATABASE_MANAGER.get_all("skills") 

294 ) 

295 skill_names = set() 

296 clean_data = [] 

297 try: 

298 for skill in skills: 

299 temp = { 

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

301 "skill_name": skill["Skill_Name"], 

302 "skill_description": skill["Skill_Description"], 

303 } 

304 

305 if ( 

306 temp["skill_name"].lower() in current_skills 

307 or temp["skill_name"].lower() in skill_names 

308 ): 

309 continue 

310 skill_names.add(temp["skill_name"].lower()) 

311 

312 temp["skill_name"] = escape(temp["skill_name"]) 

313 temp["skill_description"] = escape(temp["skill_description"]) 

314 

315 clean_data.append(temp) 

316 except Exception: 

317 return jsonify({"error": "Invalid file"}), 400 

318 

319 DATABASE_MANAGER.insert_many("skills", clean_data) 

320 

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

322 

323 def delete_all_skills(self): 

324 """Delete all skills""" 

325 from app import DATABASE_MANAGER 

326 

327 DATABASE_MANAGER.delete_all("skills") 

328 

329 students = DATABASE_MANAGER.get_all("students") 

330 

331 for student in students: 

332 if "skills" in student: 

333 student["skills"] = [] 

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

335 

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

337 

338 def delete_all_attempted_skill(self): 

339 """Delete all attempted skills""" 

340 from app import DATABASE_MANAGER 

341 

342 DATABASE_MANAGER.delete_all("attempted_skills") 

343 

344 students = DATABASE_MANAGER.get_all("students") 

345 

346 for student in students: 

347 if "attempted_skills" in student: 

348 student["attempted_skills"] = [] 

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

350 

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