None.cursor()

class MySQLController(object):
    """
        Python Class for connecting  with MySQL servers and querying,updating, or deleting using MySQL
    """

    __instance = None
    __session = None
    __connection = None
    __secret_name = None
    __config = None
    __logger = None

    def __init__(self, secret_name):
        warnings.simplefilter(action='ignore', category=UserWarning)
        if not secret_name:
            raise Exception(
                '\n Please select a Secret Name from Prefect Block!')
        else:
            self.__secret_name = secret_name
            self.__auth()

    # End def __init__

    def __auth(self):
        aws_credentials_block = AwsCredentials.load("scheduler-user")
        aws_secret = read_secret(self.__secret_name.value, aws_credentials_block)
        auth = json.loads(aws_secret)

        if 'username' in auth.keys():
            auth['user'] = auth['username']
            del auth['username']
        if 'port' in auth.keys():
            auth['port'] = int(auth['port'])
        if 'dbname' in auth.keys():
            auth['database'] = auth['dbname']
            del auth['dbname']
        if 'engine' in auth.keys():
            del auth['engine']
        if 'engin' in auth.keys():
            del auth['engin']

        auth['autocommit'] = 'true'
        auth['local_infile'] = 'true'
        self.__config = auth

    # End def __auth
    def __open(self):
        try:
            if self.__connection:
                self.__connection.ping(reconnect=True)
                self.__session = self.__connection.cursor()
            else:
                cnx = Connect(**self.__config)
                self.__connection = cnx
                self.__session = cnx.cursor()
        except MySQLError as e:
            print("Error %d: %s" % (e.args[0], e.args[1]))

    # End def __open

    def __close(self):
        try:
            if self.__connection:
                if self.__connection.open:
                    self.__connection.close()
            else:
                print('No connection to close.')
        except MySQLError as e:
            print("Error %d: %s" % (e.args[0], e.args[1]))
def query(self, select="*", db=None, table=None, where=None, *args, **kwargs):

    strSELECT = "SELECT "
    strWHERE = " "

    lstVals = []
    if not type(select) == list:
        for val in select.split(","):
            lstVals.append(val.strip(" "))
        select = lstVals

    strSELECT += ", ".join(select)
    strFROM = f" FROM {db}.{table} "
    if where:
        if not type(where) == list:
            where = [where]
        for constraint in tuple(where):
            strWHERE += f" AND {constraint}"

    query = f"""
            {strSELECT}
            {strFROM}
            WHERE 1=1
            {strWHERE}
        """

    self.__open()
    if kwargs:
        result = pd.read_sql_query(query, self.__connection, **kwargs)
    else:
        result = pd.read_sql_query(query, self.__connection)
    self.__close()
    return result

MySQLController.__open

Connect

DatabaseCredentials