db.py - master - Artist/fin_bot

db.py
70 строк2 KB
Artist
First commit
01 мар 2024, 15:48
01 мар 2024, 15:48ba257b9
import os
from typing import Dict, List, Tuple

import sqlite3

conn = sqlite3.connect(os.path.join("db", "/data/finance.db"))
cursor = conn.cursor()


def insert(table: str, column_values: Dict):
    columns = ', '.join(column_values.keys())
    values = [tuple(column_values.values())]
    placeholders = ', '.join("?" * len(column_values.keys()))
    cursor.executemany(
        f"INSERT INTO {table} "
        f"({columns}) "
        f"VALUES ({placeholders})",
        values)
    conn.commit()


def delete(table: str, row_id: int) -> None:
    row_id = int(row_id)
    cursor.execute(f"DELETE FROM {table} WHERE id={row_id}")
    conn.commit()


def update_one_row(table: str, column_values: Dict) -> None:
    column = ', '.join(column_values.keys())
    value = [list(column_values.values())][0][0]
    print(f"UPDATE {table} SET {column} = {value}")
    cursor.execute(f"UPDATE {table} SET {column} = {value}")
    conn.commit()


def fetchall(table: str, columns: List[str]) -> List[Tuple]:
    columns_joined = ", ".join(columns)
    cursor.execute(f"SELECT {columns_joined} FROM {table}")
    rows = cursor.fetchall()
    result = []
    for row in rows:
        dict_row = {}
        for index, column in enumerate(columns):
            dict_row[column] = row[index]
        result.append(dict_row)
    return result


def get_cursor():
    return cursor


def _init_db():
   """Инициализация БД"""
   with open("createdb.sql", "r") as f:
        sql = f.read()
   cursor.executescript(sql)
   conn.commit()


def check_db_exists():
    """Проверка - инициализарованна ли БД, если нет то - инициализирует!"""
    cursor.execute("SELECT name FROM sqlite_master "
                   "WHERE type='table' AND name='expense'")
    table_exists = cursor.fetchall()
    if table_exists:
        return
    _init_db()

check_db_exists()