Connect to a AWS RDS mySQL database example

Dear all,
does anyone have a code example of a Streamlit app connecting to a (AWS RDS) mySQL database and fetching data ?
Thanks.

Hi @blob123 -

Our documentation highlights an example of using a database connection with Streamlit:

https://docs.streamlit.io/en/stable/caching.html#example-1-pass-a-database-connection-around

You can connect to MySQL using the following:

https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.IAMDBAuth.Connecting.Python.html#UsingWithRDS.IAMDBAuth.Connecting.Python.AuthToken.Connect

Best,
Randy

Thanks Randy.
I was trying to use this code snippet Connect to a database - #9 by blob123 in the first place … but it didn’t work either.
Actually I have issues running queries on mySQL using a AWS RDS DB from Streamlit… if anyone has experienced it already.

I tried this way Connecting to your DB instance using IAM authentication and the AWS SDK for Python (Boto3) - Amazon Relational Database Service as well, but it failed to connect.

This is my code.

I get this error message now: TypeError: mySQL() takes no arguments

My class is not properly called … any suggestion ?

import streamlit as st
import mysql.connector
from mysql.connector import errorcode

config = {
'user': 'XXXX',
'password':'YYYY',
'host': 'DATABASE.c8tbolmoipof.us-east-2.rds.amazonaws.com',
'database':'DATABASE',
'raise_on_warnings': True
}

class mySQL:
    def init(self, st):
        self.st = st
        try:
            self.cnx = mysql.connector.connect(**config)
            self.cursor = self.cnx.cursor()
        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                self.st.text("Something is wrong with your user name or password")
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                self.st.text("Database does not exist")
            else:
                self.st.text('Unknown error')
        
    def mysql_select(self, sql) :
        try :
            self.cursor.execute(sql)
            result = self.cursor.fetchall()
            if not result :
                return False, "Record not found"
            else :
                return True, result
        except mysql.connector.Error as err:
            return False, str(err.errno) + " : " + sql

query_to_execute = """SELECT * FROM TABLE"""
mysql = mySQL(st)
mysql. mysql_select(query_to_execute)