MySQL

SQL functions to aid in MariaDB and MySQL (primarily tested on MariaDB)

Code

Source code in linecaspy/helpers/mysql.py

class MysqlFunctions:
    def get_mysql_connection(self, config):
        """Sets up MariaDB authentication

        Args:
            config (dict): of get_whisk_collector_config function

        Returns:
            pymysql.connections.Connection: object
        """
        mysql_conn = pymysql.connect(
            host=config["mysql_hostname"],
            user=config["mysql_username"],
            password=config["mysql_password"],
            db=config["mysql_database"],
        )
        logger.debug(f"mysql_conn is of type: {type(mysql_conn)} with data: {mysql_conn}")
        logger.info("Connected to mysql successfully!")
        return mysql_conn

    def mysql_insert_statement(self, mysql_conn, statement):
        """MySQL insert function

        Args:
            mysql_conn (pymysql.connections.Connection): MySQL Connection
            statement (string): SQL INSERT statement
        """
        logger.debug(f"STATEMENT: {statement}")
        if statement[:6].upper() != "INSERT":
            raise ValueError("Unsafe statement for this function.")
        cursor = mysql_conn.cursor()
        cursor.execute(statement)
        mysql_conn.commit()

    def mysql_select_statement(self, mysql_conn, statement="SELECT * FROM site"):
        """MySQL select statement for validating connections

        Args:
            mysql_conn (pymysql.connections.Connection): MySQL Connection
            statement (string): SQL SELECT statement
        """
        logger.info(f"STATEMENT: {statement}")
        if statement[:6].upper() != "SELECT":
            raise ValueError("Unsafe statement for this function.")
        cursor = mysql_conn.cursor()
        cursor.execute(statement)
        results = cursor.fetchall()
        for record in results:
            logger.info(f"RECORD: {record}")

get_mysql_connection(self, config)

Sets up MariaDB authentication

Parameters:

Name	Type	Description	Default
config	dict	of get_whisk_collector_config function	required

Returns:

Type	Description
pymysql.connections.Connection	object

Source code in linecaspy/helpers/mysql.py

def get_mysql_connection(self, config):
    """Sets up MariaDB authentication

    Args:
        config (dict): of get_whisk_collector_config function

    Returns:
        pymysql.connections.Connection: object
    """
    mysql_conn = pymysql.connect(
        host=config["mysql_hostname"],
        user=config["mysql_username"],
        password=config["mysql_password"],
        db=config["mysql_database"],
    )
    logger.debug(f"mysql_conn is of type: {type(mysql_conn)} with data: {mysql_conn}")
    logger.info("Connected to mysql successfully!")
    return mysql_conn

mysql_insert_statement(self, mysql_conn, statement)

MySQL insert function

Parameters:

Name	Type	Description	Default
mysql_conn	pymysql.connections.Connection	MySQL Connection	required
statement	string	SQL INSERT statement	required

Source code in linecaspy/helpers/mysql.py

def mysql_insert_statement(self, mysql_conn, statement):
    """MySQL insert function

    Args:
        mysql_conn (pymysql.connections.Connection): MySQL Connection
        statement (string): SQL INSERT statement
    """
    logger.debug(f"STATEMENT: {statement}")
    if statement[:6].upper() != "INSERT":
        raise ValueError("Unsafe statement for this function.")
    cursor = mysql_conn.cursor()
    cursor.execute(statement)
    mysql_conn.commit()

mysql_select_statement(self, mysql_conn, statement='SELECT * FROM site')

MySQL select statement for validating connections

Parameters:

Name	Type	Description	Default
mysql_conn	pymysql.connections.Connection	MySQL Connection	required
statement	string	SQL SELECT statement	'SELECT * FROM site'

Source code in linecaspy/helpers/mysql.py

def mysql_select_statement(self, mysql_conn, statement="SELECT * FROM site"):
    """MySQL select statement for validating connections

    Args:
        mysql_conn (pymysql.connections.Connection): MySQL Connection
        statement (string): SQL SELECT statement
    """
    logger.info(f"STATEMENT: {statement}")
    if statement[:6].upper() != "SELECT":
        raise ValueError("Unsafe statement for this function.")
    cursor = mysql_conn.cursor()
    cursor.execute(statement)
    results = cursor.fetchall()
    for record in results:
        logger.info(f"RECORD: {record}")