(self)
| 246 | t.start() |
| 247 | |
| 248 | def dataStore(self): |
| 249 | # stime = time.time() |
| 250 | cnt = 0 |
| 251 | conn = sqlite3.connect(self.dataBase, check_same_thread=False) |
| 252 | cur = conn.cursor() |
| 253 | while cnt < 1000 and not self.dataQue.empty(): |
| 254 | tableName, data = self.dataQue.get() |
| 255 | sql = 'INSERT INTO %s VALUES ' % tableName + \ |
| 256 | '(' + '?,'*(len(data)-1) + '?' + ')' |
| 257 | try: |
| 258 | cur.execute(sql, data) |
| 259 | except sqlite3.IntegrityError: |
| 260 | print("import error") |
| 261 | print(data) |
| 262 | pass |
| 263 | cnt += 1 |
| 264 | |
| 265 | conn.commit() |
| 266 | cur.close() |
| 267 | conn.close() |
| 268 | |
| 269 | self.createTimer() |
| 270 | |
| 271 | # DEFAULT_VEHTYPE |
| 272 | def getAllvTypeID(self) -> list: |
nothing calls this directly
no test coverage detected