1 """Database management."""
2 from os.path import isfile
3 from sqlite3 import connect as sql_connect
4 from plomtask.misc import HandledException
6 PATH_DB_SCHEMA = 'scripts/init.sql'
10 """Represents the sqlite3 database's file."""
12 def __init__(self, path):
17 """Check file exists and is of proper schema."""
18 self.exists = isfile(self.path)
20 self.validate_schema()
23 """Create tables in self.path file as per PATH_DB_SCHEMA sql file."""
24 with sql_connect(self.path) as conn:
25 with open(PATH_DB_SCHEMA, 'r', encoding='utf-8') as f:
26 conn.executescript(f.read())
29 def validate_schema(self):
30 """Compare found schema with what's stored at PATH_DB_SCHEMA."""
31 sql_for_schema = 'SELECT sql FROM sqlite_master ORDER BY sql'
32 msg_wrong_schema = 'Database has wrong tables schema.'
33 with sql_connect(self.path) as conn:
34 schema_rows = [r[0] for r in conn.execute(sql_for_schema) if r[0]]
35 retrieved_schema = ';\n'.join(schema_rows) + ';'
36 with open(PATH_DB_SCHEMA, 'r', encoding='utf-8') as f:
37 stored_schema = f.read().rstrip()
38 if stored_schema != retrieved_schema:
39 raise HandledException(msg_wrong_schema)