home
·
contact
·
privacy
projects
/
plomtask
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (from parent 1:
e49ef88
)
Check for database file's DB version (via user_version PRAGMA).
author
Christian Heller
<c.heller@plomlompom.de>
Wed, 24 Apr 2024 23:52:46 +0000
(
01:52
+0200)
committer
Christian Heller
<c.heller@plomlompom.de>
Wed, 24 Apr 2024 23:52:46 +0000
(
01:52
+0200)
plomtask/db.py
patch
|
blob
|
history
diff --git
a/plomtask/db.py
b/plomtask/db.py
index 848e750f1a3feccc3d9fb75c192c47a0a4edd4de..dd2ee2452a7480878c47b38f2669c08cee229ca6 100644
(file)
--- a/
plomtask/db.py
+++ b/
plomtask/db.py
@@
-6,6
+6,7
@@
from typing import Any, Dict
from plomtask.exceptions import HandledException
PATH_DB_SCHEMA = 'scripts/init.sql'
from plomtask.exceptions import HandledException
PATH_DB_SCHEMA = 'scripts/init.sql'
+EXPECTED_DB_VERSION = 0
class DatabaseFile: # pylint: disable=too-few-public-methods
class DatabaseFile: # pylint: disable=too-few-public-methods
@@
-23,11
+24,22
@@
class DatabaseFile: # pylint: disable=too-few-public-methods
self._check()
def _check(self) -> None:
self._check()
def _check(self) -> None:
- """Check file exists, and is of proper schema."""
+ """Check file exists, and is of proper
DB version and
schema."""
self.exists = isfile(self.path)
if self.exists:
self.exists = isfile(self.path)
if self.exists:
+ self._validate_user_version()
self._validate_schema()
self._validate_schema()
+ def _validate_user_version(self) -> None:
+ """Compare DB user_version with EXPECTED_DB_VERSION."""
+ sql_for_db_version = 'PRAGMA user_version'
+ with sql_connect(self.path) as conn:
+ db_version = list(conn.execute(sql_for_db_version))[0][0]
+ if db_version != EXPECTED_DB_VERSION:
+ msg = f'Wrong DB version, expected '\
+ f'{EXPECTED_DB_VERSION}, got {db_version}.'
+ raise HandledException(msg)
+
def _validate_schema(self) -> None:
"""Compare found schema with what's stored at PATH_DB_SCHEMA."""
sql_for_schema = 'SELECT sql FROM sqlite_master ORDER BY sql'
def _validate_schema(self) -> None:
"""Compare found schema with what's stored at PATH_DB_SCHEMA."""
sql_for_schema = 'SELECT sql FROM sqlite_master ORDER BY sql'