73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
from core.Constants import Constants
|
|
import sqlite3
|
|
|
|
|
|
class Model:
|
|
|
|
@staticmethod
|
|
def _create_table_if_not_exists(table_name: str, table_definition: str, drop_existing: bool = False):
|
|
|
|
connection = sqlite3.connect(Constants.HV_STORAGE_DATABASE_PATH)
|
|
cursor = connection.cursor()
|
|
|
|
if drop_existing:
|
|
cursor.execute(f'DROP TABLE IF EXISTS {table_name}')
|
|
|
|
cursor.execute(f'CREATE TABLE IF NOT EXISTS {table_name} ({table_definition})')
|
|
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _query_one(query: str, row_factory, parameters=None):
|
|
|
|
if parameters is None:
|
|
parameters = []
|
|
|
|
connection = sqlite3.connect(Constants.HV_STORAGE_DATABASE_PATH)
|
|
cursor = connection.cursor()
|
|
|
|
cursor.row_factory = row_factory
|
|
|
|
results = cursor.execute(query, parameters).fetchone()
|
|
connection.close()
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def _query_exists(query: str, parameters):
|
|
|
|
connection = sqlite3.connect(Constants.HV_STORAGE_DATABASE_PATH)
|
|
cursor = connection.cursor()
|
|
|
|
response = cursor.execute(f'SELECT EXISTS({query})', parameters).fetchone()
|
|
connection.close()
|
|
|
|
return response[0] == 1
|
|
|
|
@staticmethod
|
|
def _query_all(query: str, row_factory, parameters=None):
|
|
|
|
if parameters is None:
|
|
parameters = []
|
|
|
|
connection = sqlite3.connect(Constants.HV_STORAGE_DATABASE_PATH)
|
|
cursor = connection.cursor()
|
|
|
|
cursor.row_factory = row_factory
|
|
|
|
results = cursor.execute(query, parameters).fetchall()
|
|
connection.close()
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def _insert_many(query: str, tuple_factory, items):
|
|
|
|
connection = sqlite3.connect(Constants.HV_STORAGE_DATABASE_PATH)
|
|
cursor = connection.cursor()
|
|
|
|
cursor.executemany(query, map(tuple_factory, items))
|
|
|
|
connection.commit()
|
|
connection.close()
|