Update

tgbot/database/db_helper.py

@@ -0,0 +1,88 @@
+# - *- coding: utf- 8 - *-
+import sqlite3
+
+from tgbot.data.config import PATH_DATABASE
+from tgbot.utils.const_functions import ded
+
+
+# Преобразование полученного списка в словарь
+def dict_factory(cursor, row) -> dict:
+    save_dict = {}
+
+    for idx, col in enumerate(cursor.description):
+        save_dict[col[0]] = row[idx]
+
+    return save_dict
+
+
+# Форматирование запроса без аргументов
+def update_format(sql, parameters: dict) -> tuple[str, list]:
+    values = ", ".join([
+        f"{item} = ?" for item in parameters
+    ])
+    sql += f" {values}"
+
+    return sql, list(parameters.values())
+
+
+# Форматирование запроса с аргументами
+def update_format_where(sql, parameters: dict) -> tuple[str, list]:
+    sql += " WHERE "
+
+    sql += " AND ".join([
+        f"{item} = ?" for item in parameters
+    ])
+
+    return sql, list(parameters.values())
+
+
+################################################################################
+# Создание всех таблиц для БД
+def create_dbx():
+    with sqlite3.connect(PATH_DATABASE) as con:
+        con.row_factory = dict_factory
+
+        ############################################################
+        # Создание таблицы с хранением - пользователей
+        if len(con.execute("PRAGMA table_info(storage_users)").fetchall()) == 7:
+            print("DB was found(1/2)")
+        else:
+            con.execute(
+                ded(f"""
+                    CREATE TABLE storage_users(
+                        increment INTEGER PRIMARY KEY AUTOINCREMENT,
+                        user_id INTEGER,
+                        user_login TEXT,
+                        user_name TEXT,
+                        user_surname TEXT,
+                        user_fullname TEXT,
+                        user_unix INTEGER
+                    )
+                """)
+            )
+            print("DB was not found(1/2) | Creating...")
+
+        # Создание таблицы с хранением - настроек
+        if len(con.execute("PRAGMA table_info(storage_settings)").fetchall()) == 1:
+            print("DB was found(2/2)")
+        else:
+            con.execute(
+                ded(f"""
+                    CREATE TABLE storage_settings(
+                        status_work TEXT
+                    )
+                """)
+            )
+
+            con.execute(
+                ded(f"""
+                    INSERT INTO storage_settings(
+                        status_work
+                    )
+                    VALUES (?)
+                """),
+                [
+                    'True',
+                ]
+            )
+            print("DB was not found(2/2) | Creating...")

tgbot/database/db_settings.py

@@ -0,0 +1,36 @@
+# - *- coding: utf- 8 - *-
+import sqlite3
+
+from pydantic import BaseModel
+
+from tgbot.data.config import PATH_DATABASE
+from tgbot.database.db_helper import dict_factory, update_format
+
+
+# Модель таблицы
+class SettingsModel(BaseModel):
+    status_work: str
+
+
+# Работа с настройками
+class Settingsx:
+    storage_name = "storage_settings"
+
+    # Получение записи
+    @staticmethod
+    def get() -> SettingsModel:
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"SELECT * FROM {Settingsx.storage_name}"
+
+            return SettingsModel(**con.execute(sql).fetchone())
+
+    # Редактирование записи
+    @staticmethod
+    def update(**kwargs):
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"UPDATE {Settingsx.storage_name} SET"
+            sql, parameters = update_format(sql, kwargs)
+
+            con.execute(sql, parameters)

tgbot/database/db_users.py

@@ -0,0 +1,133 @@
+# - *- coding: utf- 8 - *-
+import sqlite3
+
+from pydantic import BaseModel
+
+from tgbot.data.config import PATH_DATABASE
+from tgbot.database.db_helper import dict_factory, update_format_where, update_format
+from tgbot.utils.const_functions import get_unix, ded
+
+
+# Модель таблицы
+class UserModel(BaseModel):
+    increment: int
+    user_id: int
+    user_login: str
+    user_name: str
+    user_surname: str
+    user_fullname: str
+    user_unix: int
+
+
+# Работа с юзером
+class Userx:
+    storage_name = "storage_users"
+
+    # Добавление записи
+    @staticmethod
+    def add(
+            user_id: int,
+            user_login: str,
+            user_name: str,
+            user_surname: str,
+            user_fullname: str,
+    ):
+        user_unix = get_unix()
+
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+
+            con.execute(
+                ded(f"""
+                    INSERT INTO {Userx.storage_name} (
+                        user_id,
+                        user_login,
+                        user_name,
+                        user_surname,
+                        user_fullname,
+                        user_unix
+                    ) VALUES (?, ?, ?, ?, ?, ?)
+                """),
+                [
+                    user_id,
+                    user_login,
+                    user_name,
+                    user_surname,
+                    user_fullname,
+                    user_unix,
+                ],
+            )
+
+    # Получение записи
+    @staticmethod
+    def get(**kwargs) -> UserModel:
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"SELECT * FROM {Userx.storage_name}"
+            sql, parameters = update_format_where(sql, kwargs)
+
+            response = con.execute(sql, parameters).fetchone()
+
+            if response is not None:
+                response = UserModel(**response)
+
+            return response
+
+    # Получение записей
+    @staticmethod
+    def gets(**kwargs) -> list[UserModel]:
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"SELECT * FROM {Userx.storage_name}"
+            sql, parameters = update_format_where(sql, kwargs)
+
+            response = con.execute(sql, parameters).fetchall()
+
+            if len(response) >= 1:
+                response = [UserModel(**cache_object) for cache_object in response]
+
+            return response
+
+    # Получение всех записей
+    @staticmethod
+    def get_all() -> list[UserModel]:
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"SELECT * FROM {Userx.storage_name}"
+
+            response = con.execute(sql).fetchall()
+
+            if len(response) >= 1:
+                response = [UserModel(**cache_object) for cache_object in response]
+
+            return response
+
+    # Редактирование записи
+    @staticmethod
+    def update(user_id, **kwargs):
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"UPDATE {Userx.storage_name} SET"
+            sql, parameters = update_format(sql, kwargs)
+            parameters.append(user_id)
+
+            con.execute(sql + "WHERE user_id = ?", parameters)
+
+    # Удаление записи
+    @staticmethod
+    def delete(**kwargs):
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"DELETE FROM {Userx.storage_name}"
+            sql, parameters = update_format_where(sql, kwargs)
+
+            con.execute(sql, parameters)
+
+    # Очистка всех записей
+    @staticmethod
+    def clear():
+        with sqlite3.connect(PATH_DATABASE) as con:
+            con.row_factory = dict_factory
+            sql = f"DELETE FROM {Userx.storage_name}"
+
+            con.execute(sql)