CoreLibs-PyPI-All/test-run/db_handling/sqlite_io.py

#!/usr/bin/env python3

"""
Main comment
"""

from pathlib import Path
from uuid import uuid4
import json
import sqlite3
from corelibs.debug_handling.dump_data import dump_data
from corelibs.logging_handling.log import Log, Logger
from corelibs.db_handling.sqlite_io import SQLiteIO

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_io.log'),
        log_name="SQLite IO",
        log_settings={
            "log_level_console": 'DEBUG',
            "log_level_file": 'DEBUG',
        }
    )
    db = SQLiteIO(
        log=Logger(log.get_logger_settings()),
        db_name=ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_io.db'),
        row_factory='Dict'
    )
    if db.db_connected():
        log.info(f"Connected to DB: {db.db_name}")
    if db.trigger_exists('trg_test_a_set_date_updated_on_update'):
        log.info("Trigger trg_test_a_set_date_updated_on_update exists")
    if db.table_exists('test_a'):
        log.info("Table test_a exists, dropping for clean test")
        db.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 = db.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 = db.execute_query(trigger_sql)
    log.debug(f"Create trigger result: {result}")
    result = db.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 = db.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 = db.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 = db.execute_query("SELECT * FROM test_a;")
    log.debug(f"Select data result: {dump_data(result)}")
    result = db.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 = db.execute_query(
        sql,
        (
            'Some updated text A',
            __uid,
        )
    )
    log.debug(f"Update data result: {dump_data(result)}")
    result = db.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,))
    log.debug(f"Fetch row after update result: {dump_data(result)}")

    db.db_close()

    db = SQLiteIO(
        log=Logger(log.get_logger_settings()),
        db_name=ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_io.db'),
        row_factory='Row'
    )
    result = db.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,))
    if result is not None and result is not False:
        log.debug(f"Fetch row result: {dump_data(result)} -> {dict(result)} -> {result.keys()}")
        log.debug(f"Access via index: {result[5]} -> {result['text_a']}")
        if isinstance(result, sqlite3.Row):
            log.debug('Result is sqlite3.Row as expected')


if __name__ == "__main__":
    main()

# __END__