check_subs/sqliter.py

import sqlite3


class Database:
    def __init__(self, path_to_db="data/main.db"):
        self.path_to_db = path_to_db

    @property
    def connection(self):
        return sqlite3.connect(self.path_to_db, timeout=2)

    def execute(self, sql: str, parameters: tuple = None, fetchone=False, fetchall=False, commit=False):
        if not parameters:
            parameters = tuple()

        connection = self.connection
        connection.set_trace_callback(logger)
        cursor = connection.cursor()
        data = None
        cursor.execute(sql, parameters)
        if commit:
            connection.commit()
        if fetchone:
            data = cursor.fetchone()
        if fetchall:
            data = cursor.fetchall()
        connection.close()

        return data

    def create_table_(self):
        sql = """
        CREATE TABLE Users (
        id int NOT NULL,
        username varchar(255) NULL,
        Name varchar(255) NOT NULL,
        PRIMARY KEY(id)
        ); 
        """
        self.execute(sql, commit=True)

    def add_word(self, word: str):
        sql = "INSERT INTO stop_words(word) VALUES(?);"
        parameters = (word,)
        self.execute(sql, parameters, commit=True)

    def select_all_word(self):
        sql = "SELECT * FROM stop_words"
        return self.execute(sql, fetchall=True)

    def delete_word(self, word: str):
        sql = 'DELETE FROM stop_words WHERE word=?'
        return self.execute(sql, parameters=(word,), commit=True)

    @staticmethod
    def format_args(sql, parameters: dict):
        sql += " AND".join([
            f" {item} = ? " for item in parameters
        ])
        return sql, tuple(parameters.values())

    # def add_channel(self, user_id: int, chat_id: int, channel_id: int):
    def add_channel(self, chat_id: int, channel_id: int):
        sql = "INSERT INTO Chats VALUES(?,?);"
        parameters = (chat_id, channel_id,)
        self.execute(sql, parameters, commit=True)

    def delete_channel(self, chat_id, channel_id):
        sql = 'DELETE FROM Chats WHERE chat_id = ? AND require_channel_id = ?'
        # sql, parameters = self.format_args(sql, kwargs)
        # return self.execute(sql, parameters, fetchone=True)
        params = (chat_id, channel_id,)
        return self.execute(sql, params, commit=True)

    def select_channels(self, chat_id):
        sql = f'SELECT * FROM Chats WHERE chat_id={chat_id};'
        return self.execute(sql, fetchall=True)

    def select_channels_by_user(self, user_id):
        sql = f'SELECT * FROM Chats WHERE user_id={user_id};'
        return self.execute(sql, fetchall=True)

    def select_all_chats(self):
        sql = f'SELECT * FROM Chats;'
        return self.execute(sql, fetchall=True)

    def update_chat_in_channel(self, chat_id, channel_id):
        sql = "UPDATE Chats SET chat_id=? WHERE require_channel_id=?"
        return self.execute(sql, parameters=(chat_id, channel_id), commit=True)

    def select_white_list(self):
        sql = f'SELECT * FROM white_list;'
        return self.execute(sql, fetchall=True)

    def select_white_list_by_chat(self, chat_id: int):
        sql = f'SELECT * FROM white_list WHERE chat_id=(?);'
        parameters = (chat_id,)
        return self.execute(sql, parameters, fetchall=True)

    def add_to_white_list(self, chat_id: int, user_id: int):
        sql = "INSERT INTO white_list VALUES(?, ?);"
        parameters = (chat_id, user_id,)
        self.execute(sql, parameters, commit=True)
'''

    def select_user(self, **kwargs):
        sql = 'SELECT * FROM Users WHERE '
        sql, parameters = self.format_args(sql, kwargs)
        return self.execute(sql, parameters, fetchone=True)

    def get_billing(self, **kwargs):
        sql = 'SELECT * FROM billing WHERE '
        sql, parameters = self.format_args(sql, kwargs)
        return self.execute(sql, parameters, fetchone=True)

    def get_all_billing(self, **kwargs):
        sql = 'SELECT * FROM billing'
        # sql, parameters = self.format_args(sql, kwargs)
        return self.execute(sql, fetchall=True)

    def count_users(self):
        return self.execute("SELECT COUNT(*) FROM Users;", fetchone=True)

    def update_bill_status(self, billing_status, billing_id):
        sql = "UPDATE billing SET billing_status=? WHERE billing_id=?"
        return self.execute(sql, parameters=(billing_status, billing_id), commit=True)

    # def delete_bill(self):
    # return self.execute('DELETE FROM billing WHERE billing')

    def delete_users(self):
        return self.execute('DELETE FROM Users WHERE True')

    def drop_table(self, name):
        return self.execute(f'DROP TABLE {name}')
'''


def logger(statement):
    print(f"""{'=' * 40}\nEXECUTING: {statement}\n{'=' * 40}""")