db.py

import logging
from collections.abc import Iterable

from sqlite3 import (
    connect,
    Connection,
    Error,
    Row
)

log = logging.getLogger(__name__)

def create_connection(
    db_file: str = "db.sqlite3"
) -> Connection | None: # Instead of Optional
    try:
        conn = connect(db_file)
        conn.row_factory = Row
        return conn
    
    except Error as SqliteErr:
        log.error(
            f"Can not connect to {db_file}. {SqliteErr}"
        )
    return None

def insert_data(
    conn: Connection, 
    table_name: str, 
    query: dict[str, tuple | list],
) -> None:
    try:
        cur = conn.cursor()
        columns_str = ', '.join(query.keys())
        placeholders = ', '.join(['?' for _ in range(len(query.keys()))])
        
        query_string = f"INSERT OR IGNORE INTO {table_name} ({columns_str}) VALUES ({placeholders})"
        cur.execute(query_string, list(query.values()))
        conn.commit()
    
    except Error as SqliteErr:
        log.error(
            f"Can not insert to table {table_name}. {SqliteErr}"
        )

def update_data_by_id(
    conn: Connection, 
    table_name: str, 
    query: dict[str, tuple | list],
    row_id: int | str
) -> None:
    try:
        cur = conn.cursor()
        set_values = ', '.join([f"{col} = ?" for col in query.keys()])
        values = (*query.values(), row_id)
        
        query_string = f"UPDATE {table_name} SET {set_values} WHERE id = ?"
        cur.execute(query_string, values)
        conn.commit()

    except Error as SqliteErr:
        log.error(
            f"Can not update data in table {table_name} by {row_id}. {SqliteErr}"
        )

def find_id_by_params(
    conn: Connection, 
    table_name: str, 
    search_query: dict[str, tuple | list],
) -> int | None:
    try:
        cur = conn.cursor()
        
        search_conditions = ' AND '.join([f"{col} = ?" for col in search_query.keys()])
        query = f"SELECT id FROM {table_name} WHERE {search_conditions}"
        
        cur.execute(query, list(search_query.values()))
        row = cur.fetchone()
        return row['id'] if row else None
    
    except Error as SqliteErr:
        log.error(
            f"Can not find data in table {table_name}"\
            f"by values {search_query.values()}. {SqliteErr}"
        )

def execute_query(
    conn: Connection, 
    query: str, 
    values: tuple = (),
    is_return: bool = True,
    is_one: bool = False
) -> None:
    try:
        cur = conn.cursor()
        
        if is_return:
            executed = cur.execute(query, values)
            table_values = executed.fetchone() if is_one else executed.fetchall()
            return table_values
        else:
            cur.execute(query, values)
            conn.commit()
            return 
    except Error as SqliteErr:
        log.error(
            f"Can not execute query {query}. {SqliteErr}"
        )