Coverage for data_drive/data_sql.py: 38%

1from pprint import pprint 

2import time 

3from datetime import time as time_from_datatime 

4import json 

5from sqlalchemy import ( 

6 create_engine, 

7 Column, 

8 Integer, 

9 String, 

10 Float, 

11) 

12from sqlalchemy.exc import IntegrityError 

13from sqlalchemy.orm import sessionmaker 

14from sqlalchemy.orm import declarative_base 

15 

16from src.settings import Settings, inf, lg 

17 

18class Database: 

19 Base = declarative_base() 

20 # creating an SQLlite database in a file 

21 engine = create_engine(Settings.data_base) 

22 

23 # Сcreating session for interective with database 

24 Session = sessionmaker(bind=engine) 

25 

26 # creating all tables that do not yet exist 

27 @classmethod 

28 def create_all_tables(cls): 

29 cls.Base.metadata.create_all(cls.engine) 

30 

31 

32 

33 

34class Beton_zawod(Database.Base): 

35 """base template for beton_zawod 

36 

37 Args: 

38 Database.Base (class): base class from sqlalchemy 

39 

40 Returns: 

41 str: description of this class 

42 """ 

43 __tablename__ = "beton_zawod" 

44 

45 id_event_time = Column(Float, primary_key=True) 

46 date_text = Column(String) 

47 list_data = Column(String) 

48 day = Column(Integer) 

49 status = Column(Integer) 

50 

51 def __repr__(self): 

52 return f"<BETHON(id_event_time={self.id_event_time}, DETE ={self.date_text})>" 

53 

54class Beton_odola(Database.Base): 

55 """base template for beton_odola 

56 

57 Args: 

58 Database.Base (class): base class from sqlalchemy 

59 

60 Returns: 

61 str: description of this class 

62 """ 

63 __tablename__ = "beton_odola" 

64 

65 id_event_time = Column(Float, primary_key=True) 

66 date_text = Column(String) 

67 list_data = Column(String) 

68 day = Column(Integer) 

69 status = Column(Integer) 

70 

71 def __repr__(self): 

72 return f"<BETHON(id_event_time={self.id_event_time}, DETE ={self.date_text})>" 

73 

74class Beton_zeran(Database.Base): 

75 """base template for beton_zeran 

76 

77 Args: 

78 Database.Base (class): base class from sqlalchemy 

79 

80 Returns: 

81 str: description of this class 

82 """ 

83 __tablename__ = "beton_zeran" 

84 

85 id_event_time = Column(Float, primary_key=True) 

86 date_text = Column(String) 

87 list_data = Column(String) 

88 day = Column(Integer) 

89 status = Column(Integer) 

90 

91 def __repr__(self): 

92 return f"<BETHON(id_event_time={self.id_event_time}, DETE ={self.date_text})>" 

93 

94class Beton_gora(Database.Base): 

95 """base template for beton_gora 

96 

97 Args: 

98 Database.Base (class): base class from sqlalchemy 

99 

100 Returns: 

101 str: description of this class 

102 """ 

103 __tablename__ = "beton_gora" 

104 

105 id_event_time = Column(Float, primary_key=True) 

106 date_text = Column(String) 

107 list_data = Column(String) 

108 day = Column(Integer) 

109 status = Column(Integer) 

110 

111 def __repr__(self): 

112 return f"<BETHON(id_event_time={self.id_event_time}, DETE ={self.date_text})>" 

113 

114class Lista_zawod(Database.Base): 

115 """base template for lista_zawod 

116 

117 Args: 

118 Database.Base (class): base class from sqlalchemy 

119 

120 Returns: 

121 str: description of this class 

122 """ 

123 __tablename__ = "lista_zawod" 

124 

125 id_event_time = Column(Float, primary_key=True) 

126 date_text = Column(String) 

127 list_data = Column(String) 

128 day = Column(Integer) 

129 status = Column(Integer) 

130 

131 def __repr__(self): 

132 return f"<LISTA(id_event_time={self.id_event_time}, DATE ={self.date_text})>" 

133 

134class Lista_odola(Database.Base): 

135 """base template for lista_odola 

136 

137 Args: 

138 Database.Base (class): base class from sqlalchemy 

139 

140 Returns: 

141 str: description of this class 

142 """ 

143 __tablename__ = "lista_odola" 

144 

145 id_event_time = Column(Float, primary_key=True) 

146 date_text = Column(String) 

147 list_data = Column(String) 

148 day = Column(Integer) 

149 status = Column(Integer) 

150 

151 def __repr__(self): 

152 return f"<LISTA(id_event_time={self.id_event_time}, DATE ={self.date_text})>" 

153 

154class Lista_zeran(Database.Base): 

155 """base template for lista_zeran 

156 

157 Args: 

158 Database.Base (class): base class from sqlalchemy 

159 

160 Returns: 

161 str: description of this class 

162 """ 

163 __tablename__ = "lista_zeran" 

164 

165 id_event_time = Column(Float, primary_key=True) 

166 date_text = Column(String) 

167 list_data = Column(String) 

168 day = Column(Integer) 

169 status = Column(Integer) 

170 

171 def __repr__(self): 

172 return f"<LISTA(id_event_time={self.id_event_time}, DATE ={self.date_text})>" 

173 

174class Lista_gora(Database.Base): 

175 """base template for lista_gora 

176 

177 Args: 

178 Database.Base (class): base class from sqlalchemy 

179 

180 Returns: 

181 str: description of this class 

182 """ 

183 __tablename__ = "lista_gora" 

184 

185 id_event_time = Column(Float, primary_key=True) 

186 date_text = Column(String) 

187 list_data = Column(String) 

188 day = Column(Integer) 

189 status = Column(Integer) 

190 

191 def __repr__(self): 

192 return f"<LISTA(id_event_time={self.id_event_time}, DATE ={self.date_text})>" 

193 

194 

195def record_beton(data): 

196 """Inserting data into the concrete orders information table 

197 

198 Args: 

199 data (dictionary): dictionary with elements wenz, day, lista_beton, date_of_day_text 

200 """ 

201 session = Database.Session() 

202 get_list_beton_serialize = [(item[0], item[1].isoformat(), *item[2:]) for item in data["lista_beton"]] 

203 serialized_list_beton = json.dumps(get_list_beton_serialize, default=str) 

204 try: 

205 if data["wenz"] == "zawod": 

206 beton = Beton_zawod( 

207 id_event_time=time.time(), 

208 date_text=data["date_of_day_text"], 

209 list_data=serialized_list_beton, 

210 day=data["day"], 

211 status=0, 

212 ) 

213 elif data["wenz"] == "odola": 

214 beton = Beton_odola( 

215 id_event_time=time.time(), 

216 date_text=data["date_of_day_text"], 

217 list_data=serialized_list_beton, 

218 day=data["day"], 

219 status=0, 

220 ) 

221 elif data["wenz"] == "zeran": 

222 beton = Beton_zeran( 

223 id_event_time=time.time(), 

224 date_text=data["date_of_day_text"], 

225 list_data=serialized_list_beton, 

226 day=data["day"], 

227 status=0, 

228 ) 

229 elif data["wenz"] == "gora": 

230 beton = Beton_gora( 

231 id_event_time=time.time(), 

232 date_text=data["date_of_day_text"], 

233 list_data=serialized_list_beton, 

234 day=data["day"], 

235 status=0, 

236 ) 

237 

238 session.add(beton) 

239 session.commit() 

240 except IntegrityError as e: 

241 lg("Ошибка целостности данных: возможно, дубликат ключа") 

242 session.rollback() # Отмена всех изменений в текущей транзакции 

243 

244 except Exception as e: 

245 lg("Ошибка при добавлении данных:", e) 

246 session.rollback() 

247 finally: 

248 session.close() 

249 

250 

251def record_lista(data): 

252 """ Inserting data into the list information table 

253 

254 Args: 

255 data (dictionary): dictionary with elements wenz, day, lista, date_of_day_text 

256 """ 

257 session = Database.Session() 

258 

259 get_list_serialize = [(item[0].isoformat(), item[1]) for item in data["lista"]] 

260 serialized_list = json.dumps(get_list_serialize, default=str) 

261 try: 

262 if data["wenz"] == "zawod": 

263 lista = Lista_zawod( 

264 id_event_time=time.time(), 

265 date_text=data["date_of_day_text"], 

266 list_data=serialized_list, 

267 day=data["day"], 

268 status=0, 

269 ) 

270 elif data["wenz"] == "odola": 

271 lista = Lista_odola( 

272 id_event_time=time.time(), 

273 date_text=data["date_of_day_text"], 

274 list_data=serialized_list, 

275 day=data["day"], 

276 status=0, 

277 ) 

278 elif data["wenz"] == "zeran": 

279 lista = Lista_zeran( 

280 id_event_time=time.time(), 

281 date_text=data["date_of_day_text"], 

282 list_data=serialized_list, 

283 day=data["day"], 

284 status=0, 

285 ) 

286 elif data["wenz"] == "gora": 

287 lista = Lista_gora( 

288 id_event_time=time.time(), 

289 date_text=data["date_of_day_text"], 

290 list_data=serialized_list, 

291 day=data["day"], 

292 status=0, 

293 ) 

294 

295 session.add(lista) 

296 session.commit() 

297 except IntegrityError as e: 

298 lg("Data integration error:possible duplicate key") 

299 session.rollback() # rollback all changes in the current transaction 

300 

301 except Exception as e: 

302 lg("Error adding data", e) 

303 session.rollback() 

304 finally: 

305 session.close() 

306 

307 

308def delete_records_below_threshold(threshold, base, wenz): 

309 """"Deletes all records from [base name] with id_event_time less than [threshold] 

310 

311 Args: 

312 threshold (float): Time as a float from the beginning of the epoch 

313 base_name (str): base name 

314 """ 

315 

316 if base == "beton" and wenz == "zawod": 

317 base_name = Beton_zawod 

318 elif base == "beton" and wenz == "odola": 

319 base_name = Beton_odola 

320 elif base == "beton" and wenz == "zeran": 

321 base_name = Beton_zeran 

322 elif base == "beton" and wenz == "gora": 

323 base_name = Beton_gora 

324 elif base == "lista" and wenz == "zawod": 

325 base_name = Lista_zawod 

326 elif base == "lista" and wenz == "odola": 

327 base_name = Lista_odola 

328 elif base == "lista" and wenz == "zeran": 

329 base_name = Lista_zeran 

330 elif base == "lista" and wenz == "gora": 

331 base_name = Lista_gora 

332 

333 session = Database.Session() 

334 

335 try: 

336 # select records with greaters primary key value 

337 records_to_delete = session.query(base_name).filter(base_name.id_event_time < threshold).order_by(base_name.id_event_time).all() 

338 

339 # delete selected records 

340 for record in records_to_delete: 

341 session.delete(record) 

342 

343 # confirm changes 

344 session.commit() 

345 except Exception as e: 

346 session.rollback() 

347 lg(f"An error occurred: {e}") 

348 finally: 

349 session.close() 

350 

351 

352def get_oldest_list_beton_or_lista(base, date_of_lista_text, wenz): 

353 """retriving the oldest list of concrete or schedule 

354 

355 Args: 

356 base (str): name base - beton or lista 

357 date_of_lista (str): date of request '10.02.2020' 

358 wenz (str): name of concretes plent  

359 

360 Returns: 

361 list: list of concrete oeders or schedule 

362 """ 

363 

364 if base == "beton" and wenz == "zawod": 

365 base_name = Beton_zawod 

366 elif base == "beton" and wenz == "odola": 

367 base_name = Beton_odola 

368 elif base == "beton" and wenz == "zeran": 

369 base_name = Beton_zeran 

370 elif base == "beton" and wenz == "gora": 

371 base_name = Beton_gora 

372 elif base == "lista" and wenz == "zawod": 

373 base_name = Lista_zawod 

374 elif base == "lista" and wenz == "odola": 

375 base_name = Lista_odola 

376 elif base == "lista" and wenz == "zeran": 

377 base_name = Lista_zeran 

378 elif base == "lista" and wenz == "gora": 

379 base_name = Lista_gora 

380 

381 session = Database.Session() 

382 

383 try: 

384 result = session.query(base_name.list_data).filter(base_name.date_text == date_of_lista_text).order_by(base_name.id_event_time.asc()).first() 

385 

386 if result: 

387 if base == "beton": 

388 deserialized_list = json.loads(result[0]) 

389 result_list = [(item[0], time_from_datatime.fromisoformat(item[1]), *item[2:]) for item in deserialized_list] 

390 return result_list 

391 

392 elif base == "lista": 

393 pass 

394 

395 return [] 

396 

397 finally: 

398 session.close() 

399 

400 

401def get_newest_list_beton_or_lista(base, date_of_lista_text, wenz): 

402 """retriving the nevest list of concrete or schedule 

403 

404 Args: 

405 base (str): name base - beton or lista 

406 date_of_lista (str): date of request '10.02.2020' 

407 wenz (str): name of concretes plent  

408 

409 Returns: 

410 list: list of concrete orders or schedule 

411 """ 

412 

413 if base == "beton" and wenz == "zawod": 

414 base_name = Beton_zawod 

415 elif base == "beton" and wenz == "odola": 

416 base_name = Beton_odola 

417 elif base == "beton" and wenz == "zeran": 

418 base_name = Beton_zeran 

419 elif base == "beton" and wenz == "gora": 

420 base_name = Beton_gora 

421 elif base == "lista" and wenz == "zawod": 

422 base_name = Lista_zawod 

423 elif base == "lista" and wenz == "odola": 

424 base_name = Lista_odola 

425 elif base == "lista" and wenz == "zeran": 

426 base_name = Lista_zeran 

427 elif base == "lista" and wenz == "gora": 

428 base_name = Lista_gora 

429 

430 session = Database.Session() 

431 

432 try: 

433 result = session.query(base_name.list_data).filter(base_name.date_text == date_of_lista_text).order_by(base_name.id_event_time.desc()).first() 

434 if result: 

435 

436 if base == "beton": 

437 deserialized_list = json.loads(result[0]) 

438 result_list = [(item[0], time_from_datatime.fromisoformat(item[1]), *item[2:]) for item in deserialized_list] 

439 return result_list 

440 

441 elif base == "lista": 

442 deserialized_list = json.loads(result[0]) 

443 result_list = [(time_from_datatime.fromisoformat(item[0]), item[1]) for item in deserialized_list] 

444 return result_list 

445 

446 return [] 

447 

448 finally: 

449 session.close() 

450 

451 

452if __name__ == '__main__': 

453 pass 

454 # pprint(get_newest_list_beton_or_lista('beton', '03.02.2025', Settings.wenzels[0]))