140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
"""
|
|
SQL Main wrapper test
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
import json
|
|
from corelibs.debug_handling.dump_data import dump_data
|
|
from corelibs.logging_handling.log import Log, Logger
|
|
from corelibs.db_handling.sql_main import SQLMain
|
|
|
|
SCRIPT_PATH: Path = Path(__file__).resolve().parent
|
|
ROOT_PATH: Path = SCRIPT_PATH
|
|
DATABASE_DIR: Path = Path("database")
|
|
LOG_DIR: Path = Path("log")
|
|
|
|
|
|
def main() -> None:
|
|
"""
|
|
Comment
|
|
"""
|
|
log = Log(
|
|
log_path=ROOT_PATH.joinpath(LOG_DIR, 'sqlite_main.log'),
|
|
log_name="SQLite Main",
|
|
log_settings={
|
|
"log_level_console": 'DEBUG',
|
|
"log_level_file": 'DEBUG',
|
|
}
|
|
)
|
|
sql_main = SQLMain(
|
|
log=Logger(log.get_logger_settings()),
|
|
db_ident=f"sqlite:{ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_main.db')}"
|
|
)
|
|
if sql_main.connected():
|
|
log.info("SQL Main connected successfully")
|
|
else:
|
|
log.error('SQL Main connection failed')
|
|
if sql_main.dbh is None:
|
|
log.error('SQL Main DBH instance is None')
|
|
return
|
|
|
|
if sql_main.dbh.trigger_exists('trg_test_a_set_date_updated_on_update'):
|
|
log.info("Trigger trg_test_a_set_date_updated_on_update exists")
|
|
if sql_main.dbh.table_exists('test_a'):
|
|
log.info("Table test_a exists, dropping for clean test")
|
|
sql_main.dbh.execute_query("DROP TABLE test_a;")
|
|
# create a dummy table
|
|
table_sql = """
|
|
CREATE TABLE IF NOT EXISTS test_a (
|
|
test_a_id INTEGER PRIMARY KEY,
|
|
date_created TEXT DEFAULT (strftime('%Y-%m-%d %H:%M:%f', 'now')),
|
|
date_updated TEXT,
|
|
uid TEXT NOT NULL UNIQUE,
|
|
set_current_timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
text_a TEXT,
|
|
content,
|
|
int_a INTEGER,
|
|
float_a REAL
|
|
);
|
|
"""
|
|
|
|
result = sql_main.dbh.execute_query(table_sql)
|
|
log.debug(f"Create table result: {result}")
|
|
trigger_sql = """
|
|
CREATE TRIGGER trg_test_a_set_date_updated_on_update
|
|
AFTER UPDATE ON test_a
|
|
FOR EACH ROW
|
|
WHEN OLD.date_updated IS NULL OR NEW.date_updated = OLD.date_updated
|
|
BEGIN
|
|
UPDATE test_a
|
|
SET date_updated = (strftime('%Y-%m-%d %H:%M:%f', 'now'))
|
|
WHERE test_a_id = NEW.test_a_id;
|
|
END;
|
|
"""
|
|
result = sql_main.dbh.execute_query(trigger_sql)
|
|
log.debug(f"Create trigger result: {result}")
|
|
result = sql_main.dbh.meta_data_detail('test_a')
|
|
log.debug(f"Table meta data detail: {dump_data(result)}")
|
|
# INSERT DATA
|
|
sql = """
|
|
INSERT INTO test_a (uid, text_a, content, int_a, float_a)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
RETURNING test_a_id, uid;
|
|
"""
|
|
result = sql_main.dbh.execute_query(
|
|
sql,
|
|
(
|
|
str(uuid4()),
|
|
'Some text A',
|
|
json.dumps({'foo': 'bar', 'number': 42}),
|
|
123,
|
|
123.456,
|
|
)
|
|
)
|
|
log.debug(f"[1] Insert data result: {dump_data(result)}")
|
|
__uid: str = ''
|
|
if result is not False:
|
|
# first one only of interest
|
|
result = dict(result[0])
|
|
__uid = str(result.get('uid', ''))
|
|
# second insert
|
|
result = sql_main.dbh.execute_query(
|
|
sql,
|
|
(
|
|
str(uuid4()),
|
|
'Some text A',
|
|
json.dumps({'foo': 'bar', 'number': 42}),
|
|
123,
|
|
123.456,
|
|
)
|
|
)
|
|
log.debug(f"[2] Insert data result: {dump_data(result)}")
|
|
result = sql_main.dbh.execute_query("SELECT * FROM test_a;")
|
|
log.debug(f"Select data result: {dump_data(result)}")
|
|
result = sql_main.dbh.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,))
|
|
log.debug(f"Fetch row result: {dump_data(result)}")
|
|
sql = """
|
|
UPDATE test_a
|
|
SET text_a = ?
|
|
WHERE uid = ?;
|
|
"""
|
|
result = sql_main.dbh.execute_query(
|
|
sql,
|
|
(
|
|
'Some updated text A',
|
|
__uid,
|
|
)
|
|
)
|
|
log.debug(f"Update data result: {dump_data(result)}")
|
|
result = sql_main.dbh.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,))
|
|
log.debug(f"Fetch row after update result: {dump_data(result)}")
|
|
|
|
sql_main.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
# __END__
|