python sqlite3, mySQL basic sample

mySQLstudy.py

import mysql.connector
from mysql.connector import Error
from sql_queries import MySqlQueries, MyDummyData


class MariaDB:
    def __init__(self):
        self.connection = None

    def create_connection(self, host_name, user_name, user_password, db_name):
        try:
            self.connection = mysql.connector.connect(
                host=host_name,
                user=user_name,
                passwd=user_password,
                database=db_name
            )
            print("Connection to MySQL DB successful")
            return True
        except Error as e:
            print(f"The error '{e}' occurred")
            return False

    def close(self):
        if self.connection is not None:
            self.connection.close()
            self.connection = None
        else:
            print("DB is not connected")

    def execute_query(self, query):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query)
                self.connection.commit()
                print("Query executed successfully")
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def execute_read_query(self, query):
        result = None
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query)
                result = cursor.fetchall()

            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

        return result

    def execute_query_param(self, query, param):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query, param)
                self.connection.commit()
                print("Query executed successfully")
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def create_users(self, name: str, age: int, gender: str, nationality: str):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(MySqlQueries.insert_users_query, (name, age, gender, nationality))
                self.connection.commit()
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def delete_comment(self, record_id: int):
        self.execute_query_param(MySqlQueries.delete_comments_query, (record_id,))
    # from here, functions for test ##########
    def initialize_with_dummy_data(self):
        # delete(drop) previous tables
        self.execute_query(MySqlQueries.drop_likes_table)
        self.execute_query(MySqlQueries.drop_comments_table)
        self.execute_query(MySqlQueries.drop_posts_table)
        self.execute_query(MySqlQueries.drop_users_table)

        # create tables
        self.execute_query(MySqlQueries.create_users_table)
        self.execute_query(MySqlQueries.create_posts_table)
        self.execute_query(MySqlQueries.create_comments_table)
        self.execute_query(MySqlQueries.create_likes_table)

        # insert dummy data
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.executemany(MySqlQueries.insert_users_query, MyDummyData.dummy_users)
                cursor.executemany(MySqlQueries.insert_posts_query, MyDummyData.dummy_posts)
                cursor.executemany(MySqlQueries.insert_comments_query, MyDummyData.dummy_comments)
                cursor.executemany(MySqlQueries.insert_likes_query, MyDummyData.dummy_likes)
                self.connection.commit()
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def print_all_users(self):
        if self.connection is not None:
            users = self.execute_read_query(MySqlQueries.select_all_users)
            if users is not None:
                for user in users:
                    print(user)
            else:
                print("select users failed")
        else:
            print("DB is not connected")

    def update_post(self):
        if self.connection is not None:
            self.execute_query_param(
                MySqlQueries.update_post_description,
                ("It's my custom post!!!", 2)
            )
        else:
            print("DB is not connected")


def main():
    db = MariaDB()
    if db.create_connection("localhost", "bato", "1234", "study"):
        db.initialize_with_dummy_data()
        #db.print_all_users()
        db.update_post()
        db.delete_comment(5)
        db.close()


if __name__ == "__main__":
    main()

sql_queries.py

from dataclasses import dataclass


@dataclass
class MyDummyData:
    dummy_users = [
        ("James", 25, "male", "USA"),
        ("Leila", 32, "female", "France"),
        ("Brigitte", 35, "female", "England"),
        ("Mike", 40, "male", "Denmark"),
        ("Elizabeth", 21, "female", "Canada")
    ]
    dummy_posts = [
        ("Happy", "I am feeling very happy today", 1),
        ("Hot Weather", "The weather is very hot today", 2),
        ("Help", "I need some help with my work", 2),
        ("Great News", "I am getting married", 1),
        ("Interesting Game", "It was a fantastic game of tennis", 5),
        ("Party", "Anyone up for a late-night party today?", 3)
    ]
    dummy_comments = [
        ("Count me In", 1, 6),
        ("What sort of help?", 5, 3),
        ("Congrats buddy", 2, 4),
        ("I was rooting for Nadal though", 4, 5),
        ("Help with your thesis?", 2, 3),
        ("Many congratulations", 5, 4)
    ]
    dummy_likes = [
        (1, 6),
        (2, 3),
        (1, 5),
        (5, 4),
        (2, 4),
        (4, 2),
        (3, 6)
    ]


@dataclass
class Sqlite3Queries:
    create_users_table = """
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        gender TEXT,
        nationality TEXT
    );
    """
    create_posts_table = """
    CREATE TABLE IF NOT EXISTS posts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        description TEXT NOT NULL,
        user_id INTEGER NOT NULL,
        FOREIGN KEY (user_id) REFERENCES users (id)
    );
    """
    create_comments_table = """
    CREATE TABLE IF NOT EXISTS comments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        text TEXT NOT NULL,
        user_id INTEGER NOT NULL,
        post_id INTEGER NOT NULL,
        FOREIGN KEY (user_id) REFERENCES users (id),
         FOREIGN KEY (post_id) REFERENCES posts (id)
    );
    """
    create_likes_table = """
    CREATE TABLE IF NOT EXISTS likes (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id INTEGER NOT NULL,
        post_id INTEGER NOT NULL,
        FOREIGN KEY (user_id) REFERENCES users (id),
        FOREIGN KEY (post_id) REFERENCES posts (id)
    );
    """
    drop_users_table = """
        DROP TABLE users;
        """
    drop_posts_table = """
        DROP TABLE posts;
        """
    drop_comments_table = """
        DROP TABLE comments;
        """
    drop_likes_table = """
        DROP TABLE likes;
        """
    insert_users_query = """
        INSERT INTO users (name, age, gender, nationality) 
        VALUES (?, ?, ?, ?);
        """
    insert_posts_query = """
        INSERT INTO posts (title, description, user_id) 
        VALUES (?, ?, ?);
        """
    insert_comments_query = """
        INSERT INTO comments (text, user_id, post_id) 
        VALUES (?, ?, ?);
        """
    insert_likes_query = """
        INSERT INTO likes (user_id, post_id) 
        VALUES (?, ?);
        """
    select_all_users = """
        SELECT * from users;
        """
    update_post_description = """
        UPDATE posts 
        SET 
            description = ?
        WHERE
            id = ? ;
        """
    delete_comments_query = """
        DELETE FROM comments 
        WHERE id = ? ;
        """


@dataclass
class MySqlQueries:
    create_users_table = """
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER  AUTO_INCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        gender TEXT,
        nationality TEXT,
        PRIMARY KEY (id)
    );
    """
    create_posts_table = """
    CREATE TABLE IF NOT EXISTS posts (
        id INTEGER AUTO_INCREMENT,
        title TEXT NOT NULL,
        description TEXT NOT NULL,
        user_id INTEGER NOT NULL,
        FOREIGN KEY fk_posts_user_id (user_id) REFERENCES users(id),
        PRIMARY KEY (id)
    );
    """
    create_comments_table = """
    CREATE TABLE IF NOT EXISTS comments(
        id INTEGER AUTO_INCREMENT,
        text TEXT NOT NULL,
        user_id INTEGER NOT NULL,
        post_id INTEGER NOT NULL,
        FOREIGN KEY fk_comments_user_id (user_id) REFERENCES users(id),
        FOREIGN KEY fk_comments_post_id (post_id) REFERENCES posts(id),
        PRIMARY KEY (id)
    );
    """
    create_likes_table = """
    CREATE TABLE IF NOT EXISTS likes (
        id INTEGER AUTO_INCREMENT,
        user_id INTEGER NOT NULL,
        post_id INTEGER NOT NULL,
        FOREIGN KEY fk_likes_user_id (user_id) REFERENCES users(id),
        FOREIGN KEY fi_likes_post_id (post_id) REFERENCES posts(id),
        PRIMARY KEY (id)
    );    
    """
    drop_users_table = """
            DROP TABLE users;
            """
    drop_posts_table = """
            DROP TABLE posts;
            """
    drop_comments_table = """
            DROP TABLE comments;
            """
    drop_likes_table = """
            DROP TABLE likes;
            """
    insert_users_query = """
        INSERT INTO users (name, age, gender, nationality) 
        VALUES ( %s, %s, %s, %s);
        """
    insert_posts_query = """
        INSERT INTO posts (title, description, user_id) 
        VALUES (%s, %s, %s);
        """
    insert_comments_query = """
        INSERT INTO comments (text, user_id, post_id) 
        VALUES (%s, %s, %s);
        """
    insert_likes_query = """
        INSERT INTO likes (user_id, post_id) 
        VALUES (%s, %s);
        """
    select_all_users = """
            SELECT * from users;
            """
    update_post_description = """
        UPDATE posts 
        SET 
            description = %s
        WHERE
            id = %s ;
        """
    delete_comments_query = """
            DELETE FROM comments 
            WHERE id = ? ;
            """

sqlite3study.py

import sqlite3
from sqlite3 import Error
from sql_queries import Sqlite3Queries, MyDummyData
from typing import List, Tuple


class Sqlite3DB:
    def __init__(self):
        self.connection = None
        self.db_path = None

    def create_connection(self, path):
        try:
            self.connection = sqlite3.connect(path)
            print("Connection to SQLite DB successful")
            return True
        except Error as e:
            print(f"The error '{e}' occurred")
            return False

    def close(self):
        if self.connection is not None:
            self.connection.close()
            self.connection = None
        else:
            print("DB is not connected")

    def execute_query(self, query):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query)
                self.connection.commit()
                print("Query executed successfully")
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def execute_query_param(self, query, param):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query, param)
                self.connection.commit()
                print("Query executed successfully")
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def execute_read_query(self, query):
        result = None
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(query)
                result = cursor.fetchall()

            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

        return result

    def delete_comment(self, record_id: int):
        self.execute_query_param(Sqlite3Queries.delete_comments_query, (record_id,))

    def create_user(self, name: str, age: int, gender: str, nationality: str):
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.execute(Sqlite3Queries.insert_users_query, (name, age, gender, nationality))
                self.connection.commit()
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    # from here, functions for test ##########
    def initialize_with_dummy_data(self):
        # delete(drop) previous tables
        self.execute_query(Sqlite3Queries.drop_users_table)
        self.execute_query(Sqlite3Queries.drop_posts_table)
        self.execute_query(Sqlite3Queries.drop_comments_table)
        self.execute_query(Sqlite3Queries.drop_likes_table)

        # create tables
        self.execute_query(Sqlite3Queries.create_users_table)
        self.execute_query(Sqlite3Queries.create_posts_table)
        self.execute_query(Sqlite3Queries.create_comments_table)
        self.execute_query(Sqlite3Queries.create_likes_table)

        # insert dummy data
        if self.connection is not None:
            cursor = self.connection.cursor()
            try:
                cursor.executemany(Sqlite3Queries.insert_users_query, MyDummyData.dummy_users)
                cursor.executemany(Sqlite3Queries.insert_posts_query, MyDummyData.dummy_posts)
                cursor.executemany(Sqlite3Queries.insert_comments_query, MyDummyData.dummy_comments)
                cursor.executemany(Sqlite3Queries.insert_likes_query, MyDummyData.dummy_likes)
                self.connection.commit()
            except Error as e:
                print(f"The error '{e}' occurred")
        else:
            print("DB is not connected")

    def print_all_users(self):
        if self.connection is not None:
            users = self.execute_read_query(Sqlite3Queries.select_all_users)
            if users is not None:
                for user in users:
                    print(user)
            else:
                print("select users failed")
        else:
            print("DB is not connected")

    def update_post(self):
        if self.connection is not None:
            self.execute_query_param(
                Sqlite3Queries.update_post_description,
                ("It's my custom post!!!", 2)
            )
        else:
            print("DB is not connected")


def main():
    db = Sqlite3DB()
    if db.create_connection("./study.sqlite"):
        db.initialize_with_dummy_data()
        #db.print_all_users()
        db.update_post()
        db.delete_comment(5)
        db.close()


if __name__ == "__main__":
    main()