database/work_db.py - patch - Esgalnor/TG_Bot

database/work_db.py
102 строки3 KB
Esgalnor
upload files
18 авг 2025, 17:24
18 авг 2025, 17:2482ee7ff
import sqlite3
import os


class WorkDB:
    """
    класс для работы с БД
    """

    def __init__(self, db_name: str):
        self.__db_name: str = db_name
        self.db_path = self.abs_path_db(self.__db_name)
        self.__tables: list[str] = self.get_list_tables()
        # self.__del_all()

    @staticmethod
    def abs_path_db(db_name: str) -> str:
        """
        возвращает полный путь до базы данных
        :param db_name: файл базы данных
        :return: полный путь
        """
        bot_path = os.path.abspath(__file__)
        bot_path = os.path.sep.join(bot_path.split(os.path.sep)[:-2])
        db_path = os.path.sep.join([bot_path, "database", db_name])
        return db_path

    def __del_table(self, table_name: str) -> None:
        """
        удаление таблицы
        :param table_name: имя удаляемой таблицы
        :return:
        """
        with sqlite3.connect(database=self.db_path) as conn:
            curs_db = conn.cursor()
            curs_db.execute(
                """
                drop table {};
                """.format(table_name)
            )

    def __del_all(self):
        """
        удаление всех таблиц
        :return:
        """
        for table in self.__tables:
            self.__del_table(table)

    def del_data(self, table_name: str) -> None:
        """
        удаление данных из таблицы
        :param table_name: название таблицы
        :return:
        """
        if table_name in self.__tables:
            with sqlite3.connect(database=self.db_path) as conn:
                curs_db = conn.cursor()
                curs_db.execute(
                    """
                    delete from {};
                    """.format(table_name)
                )

    def __str__(self):
        with sqlite3.connect(database=self.db_path) as conn:
            curs_db = conn.cursor()
            for table in self.__tables:
                curs_db.execute(
                    """
                    select * from {};
                    """.format(table)
                )
                print("\n", table)
                for string in curs_db.fetchall():
                    print(*string)
        return ""

    def get_list_tables(self) -> list:
        """
        получение списка всех таблиц в БД
        :return:
        """
        with sqlite3.connect(database=self.db_path) as conn:
            curs_db = conn.cursor()
            curs_db.execute(
                """
                                    select name from sqlite_master where type="table";
                                    """
            )
            table_list = [
                table[0]
                for table in curs_db.fetchall()
                if table[0] != "sqlite_sequence"
            ]
        return table_list


if __name__ == "__main__":
    db = WorkDB("user.db")
    # db._WorkDB__del_all()
    print('OK', db)