(self, db: sqlite3.Connection)
| 1307 | self.update_schema_version(db, latest) |
| 1308 | |
| 1309 | def load_schema(self, db: sqlite3.Connection) -> None: |
| 1310 | SQL = """ |
| 1311 | SELECT value FROM MgrModuleKV WHERE key = '__version'; |
| 1312 | """ |
| 1313 | |
| 1314 | kv = self.get_module_option('sqlite3_killpoint') |
| 1315 | with db: |
| 1316 | db.execute('BEGIN;') |
| 1317 | self.create_skeleton_schema(db) |
| 1318 | if kv == 1: |
| 1319 | return self._ceph_exit(120, hard=True) |
| 1320 | cur = db.execute(SQL) |
| 1321 | row = cur.fetchone() |
| 1322 | self.maybe_upgrade(db, int(row['value'])) |
| 1323 | assert cur.fetchone() is None |
| 1324 | cur.close() |
| 1325 | if kv == 2: |
| 1326 | return self._ceph_exit(120, hard=True) |
| 1327 | if kv == 3: |
| 1328 | return self._ceph_exit(120, hard=True) |
| 1329 | |
| 1330 | def configure_db(self, db: sqlite3.Connection) -> None: |
| 1331 | db.execute('PRAGMA FOREIGN_KEYS = 1') |
no test coverage detected