91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
from pathlib import Path
|
||
|
||
from .DatabaseAbstraction import Database, Cursor
|
||
import sqlite3 as sq
|
||
|
||
class SQLiteCursor(Cursor):
|
||
def __init__(self, cursor):
|
||
super().__init__(cursor)
|
||
self._cursor = cursor
|
||
|
||
def execute(self, sql: str, params: list = None) -> None:
|
||
"""Выполняет SQL запрос"""
|
||
if params is None:
|
||
self._cursor.execute(sql)
|
||
else:
|
||
self._cursor.execute(sql, params)
|
||
|
||
def fetchone(self, sql: str, params: list = None) -> dict:
|
||
"""Получает одну строку результата"""
|
||
if params is None:
|
||
self._cursor.execute(sql)
|
||
else:
|
||
self._cursor.execute(sql, params)
|
||
|
||
row = self._cursor.fetchone()
|
||
if row is None:
|
||
return None
|
||
|
||
# Преобразуем в словарь с именами колонок
|
||
columns = [description[0] for description in self._cursor.description]
|
||
return dict(zip(columns, row))
|
||
|
||
def fetchmany(self, sql: str = None, params: list = None) -> list[dict]:
|
||
"""Получает несколько строк результата"""
|
||
if sql is not None:
|
||
if params is None:
|
||
self._cursor.execute(sql)
|
||
else:
|
||
self._cursor.execute(sql, params)
|
||
|
||
rows = self._cursor.fetchmany()
|
||
if not rows:
|
||
return []
|
||
|
||
# Преобразуем в список словарей
|
||
columns = [description[0] for description in self._cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
def fetchall(self, sql: str, params: list = None) -> list[dict]:
|
||
"""Получает все строки результата"""
|
||
if params is None:
|
||
self._cursor.execute(sql)
|
||
else:
|
||
self._cursor.execute(sql, params)
|
||
|
||
rows = self._cursor.fetchall()
|
||
if not rows:
|
||
return []
|
||
|
||
# Преобразуем в список словарей
|
||
columns = [description[0] for description in self._cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
def lastrowid(self):
|
||
"""Возвращает ID последней вставленной строки"""
|
||
return self._cursor.lastrowid
|
||
|
||
|
||
class SQLiteDatabase(Database):
|
||
|
||
def __init__(self, name: str, path = '.'):
|
||
super().__init__(name)
|
||
self._connection: sq.Connection = sq.connect(Path(path) / (name + '.db'))
|
||
self._connection.autocommit = True
|
||
self._connection.row_factory = sq.Row # Для получения словарей
|
||
self.connected = True
|
||
|
||
def commit(self):
|
||
"""Фиксирует транзакцию"""
|
||
if self.connected and self._connection:
|
||
self._connection.commit()
|
||
|
||
def cursor(self) -> Cursor:
|
||
"""Создает и возвращает курсор"""
|
||
return SQLiteCursor(self._connection.cursor())
|
||
|
||
def close(self):
|
||
"""Закрывает соединение с базой данных"""
|
||
if self.connected and self._connection:
|
||
self._connection.close()
|
||
self.connected = False |