v0.29.0: Add SQLite IO class

This is a very basic class without many helper functions added yet
Add to the CoreLibs so when we develop it further it can be used by all projects

pyproject.toml

@@ -1,7 +1,7 @@
 # MARK: Project info
 [project]
 name = "corelibs"
-version = "0.28.0"
+version = "0.29.0"
 description = "Collection of utils for Python scripts"
 readme = "README.md"
 requires-python = ">=3.13"
@@ -63,3 +63,7 @@ ignore = [
 [tool.pylint.MASTER]
 # this is for the tests/etc folders
 init-hook='import sys; sys.path.append("src/")'
+[tool.pytest.ini_options]
+testpaths = [
+    "tests",
+]

src/corelibs/db_handling/sqlite_io.py

@@ -0,0 +1,214 @@
+"""
+SQLite DB::IO
+Will be moved to the CoreLibs
+also method names are subject to change
+"""
+
+# import gc
+from pathlib import Path
+from typing import Any, Literal, TYPE_CHECKING
+import sqlite3
+from corelibs.debug_handling.debug_helpers import call_stack
+if TYPE_CHECKING:
+    from corelibs.logging_handling.log import Logger
+
+
+class SQLiteIO():
+    """Mini SQLite interface"""
+
+    def __init__(
+        self,
+        log: 'Logger',
+        db_name: str | Path,
+        autocommit: bool = False,
+        enable_fkey: bool = True,
+        row_factory: str | None = None
+    ):
+        self.log = log
+        self.db_name = db_name
+        self.autocommit = autocommit
+        self.enable_fkey = enable_fkey
+        self.row_factory = row_factory
+        self.conn: sqlite3.Connection | None = self.db_connect()
+
+    # def __del__(self):
+    #     self.db_close()
+
+    def db_connect(self) -> sqlite3.Connection | None:
+        """
+        Connect to SQLite database, create if it doesn't exist
+        """
+        try:
+            # Connect to database (creates if doesn't exist)
+            self.conn = sqlite3.connect(self.db_name, autocommit=self.autocommit)
+            self.conn.setconfig(sqlite3.SQLITE_DBCONFIG_ENABLE_FKEY, True)
+            # self.conn.execute("PRAGMA journal_mode=WAL")
+            # self.log.debug(f"Connected to database: {self.db_name}")
+
+            def dict_factory(cursor: sqlite3.Cursor, row: list[Any]):
+                fields = [column[0] for column in cursor.description]
+                return dict(zip(fields, row))
+
+            match self.row_factory:
+                case 'Row':
+                    self.conn.row_factory = sqlite3.Row
+                case 'Dict':
+                    self.conn.row_factory = dict_factory
+                case _:
+                    self.conn.row_factory = None
+
+            return self.conn
+        except (sqlite3.Error, sqlite3.OperationalError) as e:
+            self.log.error(f"Error connecting to database [{type(e).__name__}] [{self.db_name}]: {e} [{call_stack()}]")
+            self.log.error(f"Error code: {e.sqlite_errorcode if hasattr(e, 'sqlite_errorcode') else 'N/A'}")
+            self.log.error(f"Error name: {e.sqlite_errorname if hasattr(e, 'sqlite_errorname') else 'N/A'}")
+            return None
+
+    def db_close(self):
+        """close connection"""
+        if self.conn is not None:
+            self.conn.close()
+            self.conn = None
+
+    def db_connected(self) -> bool:
+        """
+        Return True if db connection is not none
+
+        Returns:
+            bool -- _description_
+        """
+        return True if self.conn else False
+
+    def __content_exists(self, content_name: str, sql_type: str) -> bool:
+        """
+        Check if some content name for a certain type exists
+
+        Arguments:
+            content_name {str} -- _description_
+            sql_type {str} -- _description_
+
+        Returns:
+            bool -- _description_
+        """
+        if self.conn is None:
+            return False
+        try:
+            cursor = self.conn.cursor()
+            cursor.execute("""
+            SELECT name
+            FROM sqlite_master
+            WHERE type = ? AND name = ?
+            """, (sql_type, content_name,))
+            return cursor.fetchone() is not None
+        except sqlite3.Error as e:
+            self.log.error(f"Error checking table [{content_name}/{sql_type}] existence: {e} [{call_stack()}]")
+            return False
+
+    def table_exists(self, table_name: str) -> bool:
+        """
+        Check if a table exists in the database
+        """
+        return self.__content_exists(table_name, 'table')
+
+    def trigger_exists(self, trigger_name: str) -> bool:
+        """
+        Check if a triggere exits
+        """
+        return self.__content_exists(trigger_name, 'trigger')
+
+    def index_exists(self, index_name: str) -> bool:
+        """
+        Check if a triggere exits
+        """
+        return self.__content_exists(index_name, 'index')
+
+    def meta_data_detail(self, table_name: str) -> list[tuple[Any, ...]] | list[dict[str, Any]] | Literal[False]:
+        """table detail"""
+        query_show_table = """
+        SELECT
+            ti.cid, ti.name, ti.type, ti.'notnull', ti.dflt_value, ti.pk,
+            il_ii.idx_name, il_ii.idx_unique, il_ii.idx_origin, il_ii.idx_partial
+        FROM
+            sqlite_schema AS m,
+            pragma_table_info(m.name) AS ti
+            LEFT JOIN (
+                SELECT
+                    il.name AS idx_name, il.'unique' AS idx_unique, il.origin AS idx_origin, il.partial AS idx_partial,
+                    ii.cid AS tbl_cid
+                FROM
+                    sqlite_schema AS m,
+                    pragma_index_list(m.name) AS il,
+                    pragma_index_info(il.name) AS ii
+                WHERE m.name = ?1
+            ) AS il_ii ON (ti.cid = il_ii.tbl_cid)
+        WHERE
+            m.name = ?1
+        """
+        return self.execute_query(query_show_table, (table_name,))
+
+    def execute_cursor(
+        self, query: str, params: tuple[Any, ...] | None = None
+    ) -> sqlite3.Cursor | Literal[False]:
+        """execute a cursor, used in execute query or return one and for fetch_row"""
+        if self.conn is None:
+            self.log.warning(f"No connection [{call_stack()}]")
+            return False
+        try:
+            cursor = self.conn.cursor()
+            if params:
+                cursor.execute(query, params)
+            else:
+                cursor.execute(query)
+            return cursor
+        except sqlite3.Error as e:
+            self.log.error(f"Error during executing cursor [{query}:{params}]: {e} [{call_stack()}]")
+            return False
+
+    def execute_query(
+        self, query: str, params: tuple[Any, ...] | None = None
+    ) -> list[tuple[Any, ...]] | list[dict[str, Any]] | Literal[False]:
+        """query execute with or without params, returns result"""
+        if self.conn is None:
+            self.log.warning(f"No connection [{call_stack()}]")
+            return False
+        try:
+            if (cursor := self.execute_cursor(query, params)) is False:
+                return False
+            # fetch before commit because we need to get the RETURN before
+            result = cursor.fetchall()
+            # this is for INSERT/UPDATE/CREATE only
+            self.conn.commit()
+            return result
+        except sqlite3.Error as e:
+            self.log.error(f"Error during executing query [{query}:{params}]: {e} [{call_stack()}]")
+            return False
+
+    def return_one(
+            self, query: str, params: tuple[Any, ...] | None = None
+    ) -> tuple[Any, ...] | dict[str, Any] | Literal[False] | None:
+        """return one row, only for SELECT"""
+        if self.conn is None:
+            self.log.warning(f"No connection [{call_stack()}]")
+            return False
+        try:
+            if (cursor := self.execute_cursor(query, params)) is False:
+                return False
+            return cursor.fetchone()
+        except sqlite3.Error as e:
+            self.log.error(f"Error during return one: {e} [{call_stack()}]")
+            return False
+
+    def fetch_row(
+        self, cursor: sqlite3.Cursor | Literal[False]
+    ) -> tuple[Any, ...] | dict[str, Any] | Literal[False] | None:
+        """read from cursor"""
+        if self.conn is None or cursor is False:
+            self.log.warning(f"No connection [{call_stack()}]")
+            return False
+        try:
+            return cursor.fetchone()
+        except sqlite3.Error as e:
+            self.log.error(f"Error during fetch row: {e} [{call_stack()}]")
+            return False
+
+# __END__

test-run/config_handling/settings_loader.py

@@ -12,6 +12,7 @@ from corelibs.config_handling.settings_loader_handling.settings_loader_check imp
 SCRIPT_PATH: Path = Path(__file__).resolve().parent
 ROOT_PATH: Path = SCRIPT_PATH
 CONFIG_DIR: Path = Path("config")
+LOG_DIR: Path = Path("log")
 CONFIG_FILE: str = "settings.ini"
 
 
@@ -26,9 +27,8 @@ def main():
     print(f"regex {regex_c} check against {value} -> {result}")
 
     # for log testing
-    script_path: Path = Path(__file__).resolve().parent
     log = Log(
-        log_path=script_path.joinpath('log', 'settings_loader.log'),
+        log_path=ROOT_PATH.joinpath(LOG_DIR, 'settings_loader.log'),
         log_name="Settings Loader",
         log_settings={
             "log_level_console": 'DEBUG',

test-run/db_handling/database/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

test-run/db_handling/log/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

test-run/db_handling/sqlite_io.py

@@ -0,0 +1,148 @@
+#!/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__

tests/unit/db_handling/__init__.py

@@ -0,0 +1,3 @@
+"""
+db_handling tests
+"""

uv.lock

@@ -108,7 +108,7 @@ wheels = [
 
 [[package]]
 name = "corelibs"
-version = "0.28.0"
+version = "0.29.0"
 source = { editable = "." }
 dependencies = [
     { name = "cryptography" },