(self, db: sqlite3.Connection, version: int)
| 1286 | return v |
| 1287 | |
| 1288 | def maybe_upgrade(self, db: sqlite3.Connection, version: int) -> None: |
| 1289 | if version <= 0: |
| 1290 | assert hasattr(self, 'SCHEMA') |
| 1291 | self.log.info(f"creating main.db for {self.module_name}") |
| 1292 | assert self.SCHEMA is not None |
| 1293 | for sql in self.SCHEMA: |
| 1294 | db.execute(sql) |
| 1295 | self.update_schema_version(db, 1) |
| 1296 | else: |
| 1297 | assert hasattr(self, 'SCHEMA_VERSIONED') |
| 1298 | assert self.SCHEMA_VERSIONED is not None |
| 1299 | latest = len(self.SCHEMA_VERSIONED) |
| 1300 | if latest < version: |
| 1301 | raise RuntimeError(f"main.db version is newer ({version}) than module ({latest})") |
| 1302 | for i in range(version, latest): |
| 1303 | self.log.info(f"upgrading main.db for {self.module_name} from {i - 1}:{i}") |
| 1304 | for sql in self.SCHEMA_VERSIONED[i]: |
| 1305 | db.execute(sql) |
| 1306 | if version < latest: |
| 1307 | self.update_schema_version(db, latest) |
| 1308 | |
| 1309 | def load_schema(self, db: sqlite3.Connection) -> None: |
| 1310 | SQL = """ |
no test coverage detected