Connect to the database, and create tables if necessary.
(self)
| 320 | |
| 321 | @catch_corrupt_db |
| 322 | def init_db(self) -> None: |
| 323 | """Connect to the database, and create tables if necessary.""" |
| 324 | if not self.enabled: |
| 325 | self.db = DummyDB() |
| 326 | self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db) |
| 327 | return |
| 328 | |
| 329 | # use detect_types so that timestamps return datetime objects |
| 330 | kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES) |
| 331 | kwargs.update(self.connection_options) |
| 332 | self.db = sqlite3.connect(str(self.hist_file), **kwargs) # type: ignore [call-overload] |
| 333 | self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db) |
| 334 | with self.db: |
| 335 | self.db.execute( |
| 336 | """CREATE TABLE IF NOT EXISTS sessions (session integer |
| 337 | primary key autoincrement, start timestamp, |
| 338 | end timestamp, num_cmds integer, remark text)""" |
| 339 | ) |
| 340 | self.db.execute( |
| 341 | """CREATE TABLE IF NOT EXISTS history |
| 342 | (session integer, line integer, source text, source_raw text, |
| 343 | PRIMARY KEY (session, line))""" |
| 344 | ) |
| 345 | # Output history is optional, but ensure the table's there so it can be |
| 346 | # enabled later. |
| 347 | self.db.execute( |
| 348 | """CREATE TABLE IF NOT EXISTS output_history |
| 349 | (session integer, line integer, output text, |
| 350 | PRIMARY KEY (session, line))""" |
| 351 | ) |
| 352 | # success! reset corrupt db count |
| 353 | self._corrupt_db_counter = 0 |
| 354 | |
| 355 | def writeout_cache(self) -> None: |
| 356 | """Overridden by HistoryManager to dump the cache before certain |