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.SP_STORAGE_DATABASE)
|
|
cursor = connection.cursor()
|
|
|
|
if drop_existing:
|
|
cursor.execute('DROP TABLE IF EXISTS ' + table_name)
|
|
|
|
cursor.execute('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.SP_STORAGE_DATABASE)
|
|
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.SP_STORAGE_DATABASE)
|
|
cursor = connection.cursor()
|
|
|
|
response = cursor.execute('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.SP_STORAGE_DATABASE)
|
|
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.SP_STORAGE_DATABASE)
|
|
cursor = connection.cursor()
|
|
|
|
cursor.executemany(query, map(tuple_factory, items))
|
|
|
|
connection.commit()
|
|
connection.close()
|