db.py

db.py

/**
sudo mysql -u root

> CREATE USER 'sammy'@'localhost' IDENTIFIED BY 'password';
> ALTER USER 'sammy'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';
> GRANT ALL PRIVILEGES ON dbTest.* To 'user'@'hostname' IDENTIFIED BY 'password';
> FLUSH PRIVILEGES;

> mysql://deepanshu:deepanshu@localhost:3306/my_db
**/

from typing import List, Dict

# pip install mysql-connector-python
import mysql.connector

# pip install psycopg2-binary
import psycopg2

# pip install snowflake-connector-python
import snowflake.connector

class SQLHelper(object):

    CONNECTOR = mysql.connector.connect
    # CONNECTOR = psycopg2.connect

    # HOST = 'mysql.gb.stackcp.com'
    # PORT = '61486'
    # DB_NAME = "deepanshu-35303135167b"
    # USR = 'deepanshu'
    # PWD = 'mypass@'

    HOST = 'mysql.gb.stackcp.com'
    PORT = '59749'
    DB_NAME = "swati_db-3635b51d"
    USR = 'swati'
    PWD = 'mypass@'

    def __init__(self):
        self.conn = None
        self.cursor = None

    def get_db_conn(self):
        # Connect to the MySQL database
        self.conn = SQLHelper.CONNECTOR(
            host=SQLHelper.HOST,
            port=SQLHelper.PORT,
            database=SQLHelper.DB_NAME,
            user=SQLHelper.USR,
            password=SQLHelper.PWD
            # options=f"-c search_path={schema}"  # for postgres becoz, db.schema_name.table_name
        )
        self.cursor = self.conn.cursor()
        
        # SnowFlake
        # ACC.ORG.REGION.WAREHOUSE.db.schema.table
        self.conn = snowflake.connector.connect(
            user='XXXX',
            password='XXXX',
            account="",
            warehouse="",
            database="",
            schema="",
            session_parameter={},
        )
        # dict cursor of SF
        self.cursor = self.conn.cursor(snowflake.connector.DictCursor)

    def execute_and_commit(self, sql_query: str):
        self.get_db_conn()
        self.cursor.execute(sql_query)

        # committing changes
        self.conn.commit()

        # closing changes
        self.close_conn()
        # print(f"Query got executed {sql_query}")

    def get_execute_data(self, sql_query) -> List[Dict]:
        self.get_db_conn()
        self.cursor.execute(sql_query)

        # fetching data
        results = self.cursor.fetchall()
        self.close_conn()

        return results

    def close_conn(self):
        # closing connection and cursor
        self.cursor.close()
        self.conn.close()

    def bulk_insert(self, sql, sql_parameters, batch_size=1000):
        """
        :param sql: "INSERT INTO user (name, email) VALUES (%s, %s)"
        :param sql_parameters: values -->[['name1', '[email protected]'], ]
        """

        for i in range(0, len(sql_parameters), batch_size):
            batch = sql_parameters[i:i + batch_size]
            self.get_db_conn()
            self.cursor.executemany(sql, batch)
            self.conn.commit()

        print(f"New {len(sql_parameters)} Entries dumped !!")


db = SQLHelper()
# ans = db.get_execute_data("""CREATE TABLE Persons (
#     PersonID int,
#     LastName varchar(255),
#     FirstName varchar(255),
#     Address varchar(255),
#     City varchar(255)
# );""")

ans = db.get_execute_data("""SELECT * FROM Persons;""")
print(ans)