def rewrite_relations(self, table_name: str, key: str, target: int,
rows: list[list[Any]]) -> None:
"""Rewrite relations in table_name to target, with rows values."""
- self.exec(f'DELETE FROM {table_name} WHERE {key} = ?', (target,))
+ self.delete_where(table_name, key, target)
for row in rows:
values = tuple([target] + row)
q_marks = self.__class__.q_marks_from_values(values)
self.exec(f'INSERT INTO {table_name} VALUES {q_marks}', values)
+ def row_where(self, table_name: str, key: str,
+ target: int | str) -> list[Row]:
+ """Return list of Rows at table where key == target."""
+ return list(self.exec(f'SELECT * FROM {table_name} WHERE {key} = ?',
+ (target,)))
+
+ def column_where(self, table_name: str, column: str, key: str,
+ target: int | str) -> list[Any]:
+ """Return column of table where key == target."""
+ return [row[0] for row in
+ self.exec(f'SELECT {column} FROM {table_name} '
+ f'WHERE {key} = ?', (target,))]
+
+ def column_all(self, table_name: str, column: str) -> list[Any]:
+ """Return complete column of table."""
+ return [row[0] for row in
+ self.exec(f'SELECT {column} FROM {table_name}')]
+
+ def delete_where(self, table_name: str, key: str, target: int) -> None:
+ """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."""
obj = cache[id_]
from_cache = True
else:
- for row in db_conn.exec(f'SELECT * FROM {cls.table_name} '
- 'WHERE id = ?', (id_,)):
+ for row in db_conn.row_where(cls.table_name, 'id', id_):
obj = cls.from_table_row(db_conn, row)
cache[id_] = obj
break