"""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
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:
"""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()
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:
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."""
"""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."""
"""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()
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]:
"""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]')
"""
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()