Create MySQL Connection

CreateMySQLConnection

import mysql.connector
from mysql.connector import Error
import sqlalchemy


# Establish a connection to MySQL server

def create_server_connection(host_name, user_name, user_password):
    
    connection = None

    try:
        connection = mysql.connector.connect(
            host = host_name,
            user = user_name,
            passwd = user_password
        )

        print('MySQL server connected successfully.')

    except Error as e:
        print(f'Error: {e}')

    return connection


# Create a new DB on MySQL server

def create_db(connection, query):
    
    cursor = connection.cursor()

    try:
        cursor.execute(query)

        print('MySQL DB created successfully.')

    except Error as e:
        print(f'Error: {e}')


# Connect to an existing DB on MySQL server

def create_db_connection(host_name, user_name, user_password, db_name):

    connection = None

    try:
        connection = mysql.connector.connect(
            host = host_name,
            user = user_name,
            passwd = user_password,
            database = db_name
        )

        print('MySQL DB connected successfully.')
    
    except Error as e:
        print(f'Error: {e}')

    return connection


# Execute a query

def execute_sql(connection, query):
    
    cursor = connection.cursor()

    try:
        cursor.execute(query)
        
        # make sure that the commands are implemented
        connection.commit() 

        print(f'Query executed successfully.')

    except Error as e:
        print(f'Error: {e}')    


# Execute many queries

def executemany_sql(connection, query, val):
    
    cursor = connection.cursor()

    try:
        cursor.executemany(query, val)
        
        connection.commit() 

        print(f'Query executed successfully.')

    except Error as e:
        print(f'Error: {e}')    


# Execute data dump by linking sqlalchemy and pandas

def create_engine(db = '', **kwargs):
    
    if not db:
        db = 'Instagram'

    username = kwargs.get('username', '')
    pw = kwargs.get('pw', '')
    host = kwargs.get('host', '')
    port = kwargs.get('port', '')

    connection = None

    try:
        connection = sqlalchemy.create_engine(
            'mysql+mysqlconnector://{0}:{1}@{2}:{3}/{4}'.
            format(username, pw, host, port, db),
        )

        print('MySQL DB connected successfully.')
    
    except Error as e:
        print(f'Error: {e}')    
    
    return connection


# Read a query

def read_sql(connection, query):
    
    cursor = connection.cursor()

    result = None

    try:
        cursor.execute(query)

        result = cursor.fetchall()

        return result
    
    except Error as e:
        print(f'Error: {e}')