plomlompom.com Git - plomtask/blob - plomtask/db.py

   1 """Database management."""
   2 from __future__ import annotations
   3 from os.path import isfile
   4 from difflib import Differ
   5 from sqlite3 import connect as sql_connect, Cursor, Row
   6 from typing import Any, Self, TypeVar, Generic
   7 from plomtask.exceptions import HandledException, NotFoundException
   8
   9 PATH_DB_SCHEMA = 'scripts/init.sql'
  10 EXPECTED_DB_VERSION = 0
  11
  12
  13 class DatabaseFile:  # pylint: disable=too-few-public-methods
  14     """Represents the sqlite3 database's file."""
  15
  16     def __init__(self, path: str) -> None:
  17         self.path = path
  18         self._check()
  19
  20     def remake(self) -> None:
  21         """Create tables in self.path file as per PATH_DB_SCHEMA sql file."""
  22         with sql_connect(self.path) as conn:
  23             with open(PATH_DB_SCHEMA, 'r', encoding='utf-8') as f:
  24                 conn.executescript(f.read())
  25         self._check()
  26
  27     def _check(self) -> None:
  28         """Check file exists, and is of proper DB version and schema."""
  29         self.exists = isfile(self.path)
  30         if self.exists:
  31             self._validate_user_version()
  32             self._validate_schema()
  33
  34     def _validate_user_version(self) -> None:
  35         """Compare DB user_version with EXPECTED_DB_VERSION."""
  36         sql_for_db_version = 'PRAGMA user_version'
  37         with sql_connect(self.path) as conn:
  38             db_version = list(conn.execute(sql_for_db_version))[0][0]
  39             if db_version != EXPECTED_DB_VERSION:
  40                 msg = f'Wrong DB version, expected '\
  41                         f'{EXPECTED_DB_VERSION}, got {db_version}.'
  42                 raise HandledException(msg)
  43
  44     def _validate_schema(self) -> None:
  45         """Compare found schema with what's stored at PATH_DB_SCHEMA."""
  46         sql_for_schema = 'SELECT sql FROM sqlite_master ORDER BY sql'
  47         msg_err = 'Database has wrong tables schema. Diff:\n'
  48         with sql_connect(self.path) as conn:
  49             schema_rows = [r[0] for r in conn.execute(sql_for_schema) if r[0]]
  50             retrieved_schema = ';\n'.join(schema_rows) + ';'
  51             with open(PATH_DB_SCHEMA, 'r', encoding='utf-8') as f:
  52                 stored_schema = f.read().rstrip()
  53                 if stored_schema != retrieved_schema:
  54                     diff_msg = Differ().compare(retrieved_schema.splitlines(),
  55                                                 stored_schema.splitlines())
  56                     raise HandledException(msg_err + '\n'.join(diff_msg))
  57
  58
  59 class DatabaseConnection:
  60     """A single connection to the database."""
  61
  62     def __init__(self, db_file: DatabaseFile) -> None:
  63         self.file = db_file
  64         self.conn = sql_connect(self.file.path)
  65
  66     def commit(self) -> None:
  67         """Commit SQL transaction."""
  68         self.conn.commit()
  69
  70     def exec(self, code: str, inputs: tuple[Any, ...] = tuple()) -> Cursor:
  71         """Add commands to SQL transaction."""
  72         return self.conn.execute(code, inputs)
  73
  74     def close(self) -> None:
  75         """Close DB connection."""
  76         self.conn.close()
  77
  78     def rewrite_relations(self, table_name: str, key: str, target: int,
  79                           rows: list[list[Any]]) -> None:
  80         """Rewrite relations in table_name to target, with rows values."""
  81         self.delete_where(table_name, key, target)
  82         for row in rows:
  83             values = tuple([target] + row)
  84             q_marks = self.__class__.q_marks_from_values(values)
  85             self.exec(f'INSERT INTO {table_name} VALUES {q_marks}', values)
  86
  87     def row_where(self, table_name: str, key: str,
  88                   target: int | str) -> list[Row]:
  89         """Return list of Rows at table where key == target."""
  90         return list(self.exec(f'SELECT * FROM {table_name} WHERE {key} = ?',
  91                               (target,)))
  92
  93     def column_where(self, table_name: str, column: str, key: str,
  94                      target: int | str) -> list[Any]:
  95         """Return column of table where key == target."""
  96         return [row[0] for row in
  97                 self.exec(f'SELECT {column} FROM {table_name} '
  98                           f'WHERE {key} = ?', (target,))]
  99
 100     def column_all(self, table_name: str, column: str) -> list[Any]:
 101         """Return complete column of table."""
 102         return [row[0] for row in
 103                 self.exec(f'SELECT {column} FROM {table_name}')]
 104
 105     def delete_where(self, table_name: str, key: str, target: int) -> None:
 106         """Delete from table where key == target."""
 107         self.exec(f'DELETE FROM {table_name} WHERE {key} = ?', (target,))
 108
 109     @staticmethod
 110     def q_marks_from_values(values: tuple[Any]) -> str:
 111         """Return placeholder to insert values into SQL code."""
 112         return '(' + ','.join(['?'] * len(values)) + ')'
 113
 114
 115 BaseModelId = TypeVar('BaseModelId', int, str)
 116 BaseModelInstance = TypeVar('BaseModelInstance', bound='BaseModel[Any]')
 117
 118
 119 class BaseModel(Generic[BaseModelId]):
 120     """Template for most of the models we use/derive from the DB."""
 121     table_name = ''
 122     to_save: list[str] = []
 123     id_: None | BaseModelId
 124     cache_: dict[BaseModelId, Self]
 125
 126     def __init__(self, id_: BaseModelId | None) -> None:
 127         if isinstance(id_, int) and id_ < 1:
 128             msg = f'illegal {self.__class__.__name__} ID, must be >=1: {id_}'
 129             raise HandledException(msg)
 130         self.id_ = id_
 131
 132     @classmethod
 133     def get_cached(cls: type[BaseModelInstance],
 134                    id_: BaseModelId) -> BaseModelInstance | None:
 135         """Get object of id_ from class's cache, or None if not found."""
 136         # pylint: disable=consider-iterating-dictionary
 137         cache = cls.get_cache()
 138         if id_ in cache.keys():
 139             obj = cache[id_]
 140             assert isinstance(obj, cls)
 141             return obj
 142         return None
 143
 144     @classmethod
 145     def empty_cache(cls) -> None:
 146         """Empty class's cache."""
 147         cls.cache_ = {}
 148
 149     @classmethod
 150     def get_cache(cls: type[BaseModelInstance]) -> dict[Any, BaseModel[Any]]:
 151         """Get cache dictionary, create it if not yet existing."""
 152         if not hasattr(cls, 'cache_'):
 153             d: dict[Any, BaseModel[Any]] = {}
 154             cls.cache_ = d
 155         return cls.cache_
 156
 157     def cache(self) -> None:
 158         """Update object in class's cache."""
 159         if self.id_ is None:
 160             raise HandledException('Cannot cache object without ID.')
 161         cache = self.__class__.get_cache()
 162         cache[self.id_] = self
 163
 164     def uncache(self) -> None:
 165         """Remove self from cache."""
 166         if self.id_ is None:
 167             raise HandledException('Cannot un-cache object without ID.')
 168         cache = self.__class__.get_cache()
 169         del cache[self.id_]
 170
 171     @classmethod
 172     def from_table_row(cls: type[BaseModelInstance],
 173                        # pylint: disable=unused-argument
 174                        db_conn: DatabaseConnection,
 175                        row: Row | list[Any]) -> BaseModelInstance:
 176         """Make from DB row, write to DB cache."""
 177         obj = cls(*row)
 178         obj.cache()
 179         return obj
 180
 181     @classmethod
 182     def _by_id(cls, db_conn: DatabaseConnection,
 183                id_: BaseModelId) -> Self | None:
 184         """Return instance found by ID, or None, and if from cache or not."""
 185         obj = cls.get_cached(id_)
 186         if not obj:
 187             for row in db_conn.row_where(cls.table_name, 'id', id_):
 188                 obj = cls.from_table_row(db_conn, row)
 189                 obj.cache()
 190                 break
 191         return obj
 192
 193     @classmethod
 194     def by_id(cls, db_conn: DatabaseConnection,
 195               id_: BaseModelId | None,
 196               # pylint: disable=unused-argument
 197               create: bool = False) -> Self:
 198         """Retrieve by id_, on failure throw NotFoundException."""
 199         obj = None
 200         if id_ is not None:
 201             obj = cls._by_id(db_conn, id_)
 202         if obj:
 203             return obj
 204         if create:
 205             obj = cls(id_)
 206             return obj
 207         raise NotFoundException(f'found no object of ID {id_}')
 208
 209     @classmethod
 210     def all(cls: type[BaseModelInstance],
 211             db_conn: DatabaseConnection) -> list[BaseModelInstance]:
 212         """Collect all objects of class."""
 213         items: dict[BaseModelId, BaseModelInstance] = {}
 214         for k, v in cls.get_cache().items():
 215             assert isinstance(v, cls)
 216             items[k] = v
 217         already_recorded = items.keys()
 218         for id_ in db_conn.column_all(cls.table_name, 'id'):
 219             if id_ not in already_recorded:
 220                 item = cls.by_id(db_conn, id_)
 221                 assert item.id_ is not None
 222                 items[item.id_] = item
 223         return list(items.values())
 224
 225     def save_core(self, db_conn: DatabaseConnection,
 226                   update_with_lastrowid: bool = True) -> None:
 227         """Write bare-bones self (sans connected items), ensuring self.id_."""
 228         values = tuple([self.id_] + [getattr(self, key)
 229                                      for key in self.to_save])
 230         q_marks = DatabaseConnection.q_marks_from_values(values)
 231         table_name = self.table_name
 232         cursor = db_conn.exec(f'REPLACE INTO {table_name} VALUES {q_marks}',
 233                               values)
 234         if update_with_lastrowid:
 235             self.id_ = cursor.lastrowid  # type: ignore[assignment]
 236         self.cache()