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

###################################################################
POSTGRES

> sudo service postgresql status
> sudo -u postgres psql
> sudo apt install postgresql postgresql-contrib

uv pip install psycopg2-binary snowflake-connector-python mysql-connector-python

CREATE USER myuser WITH ENCRYPTED PASSWORD 'mypassword';
GRANT ALL PRIVILEGES ON DATABASE mydb TO myuser;
"""

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
from psycopg2.extras import RealDictCursor


class DBConfigs(object):
    # CONNECTOR = mysql.connector.connect
    CONNECTOR = psycopg2.connect

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

    HOST = 'localhost'
    PORT = '5432'  # '59749'
    DB_NAME = "dee"
    USR = 'deepanshu'
    PWD = 'deepanshu'

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

    def get_db_conn(self):
        db = 'psql' # 'mysql', 'snow'

        if db == 'mysql' or db == 'psql':
            # Connect to the MySQL database
            self.conn = DBConfigs.CONNECTOR(
                host=DBConfigs.HOST,
                port=DBConfigs.PORT,
                database=DBConfigs.DB_NAME,
                user=DBConfigs.USR,
                password=DBConfigs.PWD,
                # options=f"-c search_path={schema}"  # for postgres becoz, db.schema_name.table_name
            )
            if db == 'psql':
                cursor = self.conn.cursor(cursor_factory=RealDictCursor) 
            else:
                cursor = self.conn.cursor()
            return cursor

        if db == 'snow':
            # 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
            cursor = self.conn.cursor(snowflake.connector.DictCursor)
            return cursor

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

        # committing changes
        self.conn.commit()

        # closing changes
        self.conn.close()
        curr.close()

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

        # fetching data
        results = curr.fetchall()

        self.conn.close()
        curr.close()

        return results

    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@gmail.com'], ]
        """

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

        self.conn.close()
        print(f"New {len(sql_parameters)} Entries dumped !!")


db = DBConfigs()
# 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)