From: Christian Heller Date: Fri, 14 Jun 2024 18:03:37 +0000 (+0200) Subject: Refactor database management code a little bit. X-Git-Url: https://plomlompom.com/repos/%7B%7Bdb.prefix%7D%7D/%7B%7B%20web_path%20%7D%7D/decks/condition_titles?a=commitdiff_plain;h=4ee3cc797f983d41dae831cb47e2b1a5d4c96f22;p=plomtask Refactor database management code a little bit. --- diff --git a/plomtask/db.py b/plomtask/db.py index bd60968..2077d79 100644 --- a/plomtask/db.py +++ b/plomtask/db.py @@ -18,8 +18,9 @@ class UnmigratedDbException(HandledException): """To identify case of unmigrated DB file.""" -class DatabaseFile: # pylint: disable=too-few-public-methods +class DatabaseFile: """Represents the sqlite3 database's file.""" + # pylint: disable=too-few-public-methods def __init__(self, path: str) -> None: self.path = path @@ -38,7 +39,7 @@ class DatabaseFile: # pylint: disable=too-few-public-methods def migrate(cls, path: str) -> DatabaseFile: """Apply migrations from_version to EXPECTED_DB_VERSION.""" migrations = cls._available_migrations() - from_version = cls.get_version_of_db(path) + from_version = cls._get_version_of_db(path) migrations_todo = migrations[from_version+1:] for j, filename in enumerate(migrations_todo): with sql_connect(path) as conn: @@ -54,7 +55,7 @@ class DatabaseFile: # pylint: disable=too-few-public-methods """Check file exists, and is of proper DB version and schema.""" if not isfile(self.path): raise NotFoundException - if self.user_version != EXPECTED_DB_VERSION: + if self._user_version != EXPECTED_DB_VERSION: raise UnmigratedDbException() self._validate_schema() @@ -86,7 +87,7 @@ class DatabaseFile: # pylint: disable=too-few-public-methods return migrations_list @staticmethod - def get_version_of_db(path: str) -> int: + def _get_version_of_db(path: str) -> int: """Get DB user_version, fail if outside expected range.""" sql_for_db_version = 'PRAGMA user_version' with sql_connect(path) as conn: @@ -99,9 +100,11 @@ class DatabaseFile: # pylint: disable=too-few-public-methods return db_version @property - def user_version(self) -> int: + def _user_version(self) -> int: """Get DB user_version.""" - return self.__class__.get_version_of_db(self.path) + # pylint: disable=protected-access + # (since we remain within class) + return self.__class__._get_version_of_db(self.path) def _validate_schema(self) -> None: """Compare found schema with what's stored at PATH_DB_SCHEMA.""" @@ -156,8 +159,7 @@ class DatabaseConnection: """A single connection to the database.""" def __init__(self, db_file: DatabaseFile) -> None: - self.file = db_file - self.conn = sql_connect(self.file.path) + self.conn = sql_connect(db_file.path) def commit(self) -> None: """Commit SQL transaction.""" @@ -167,6 +169,11 @@ class DatabaseConnection: """Add commands to SQL transaction.""" return self.conn.execute(code, inputs) + def exec_on_vals(self, code: str, inputs: tuple[Any, ...]) -> Cursor: + """Wrapper around .exec appending adequate " (?, …)" to code.""" + q_marks_from_values = '(' + ','.join(['?'] * len(inputs)) + ')' + return self.exec(f'{code} {q_marks_from_values}', inputs) + def close(self) -> None: """Close DB connection.""" self.conn.close() @@ -183,8 +190,7 @@ class DatabaseConnection: self.delete_where(table_name, key, target) for row in rows: values = tuple(row[:key_index] + [target] + row[key_index:]) - q_marks = self.__class__.q_marks_from_values(values) - self.exec(f'INSERT INTO {table_name} VALUES {q_marks}', values) + self.exec_on_vals(f'INSERT INTO {table_name} VALUES', values) def row_where(self, table_name: str, key: str, target: int | str) -> list[Row]: @@ -220,11 +226,6 @@ class DatabaseConnection: """Delete from table where key == target.""" self.exec(f'DELETE FROM {table_name} WHERE {key} = ?', (target,)) - @staticmethod - def q_marks_from_values(values: tuple[Any]) -> str: - """Return placeholder to insert values into SQL code.""" - return '(' + ','.join(['?'] * len(values)) + ')' - BaseModelId = TypeVar('BaseModelId', int, str) BaseModelInstance = TypeVar('BaseModelInstance', bound='BaseModel[Any]') @@ -417,10 +418,9 @@ class BaseModel(Generic[BaseModelId]): """ values = tuple([self.id_] + [getattr(self, key) for key in self.to_save]) - q_marks = DatabaseConnection.q_marks_from_values(values) table_name = self.table_name - cursor = db_conn.exec(f'REPLACE INTO {table_name} VALUES {q_marks}', - values) + cursor = db_conn.exec_on_vals(f'REPLACE INTO {table_name} VALUES', + values) if not isinstance(self.id_, str): self.id_ = cursor.lastrowid # type: ignore[assignment] self.cache()