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)

@randyzwitch - it would be really good to have a worked example of getting data from a relational database:

  • credentials such as hostname, db, username and password from config file or env vars
  • instantiation of a db connection or pool such as sqlalchemy
  • how to use that pool for functions that run queries, particularly such that the connection object is correctly cached (ie: not cached :wink: but the results of the function generating the query are.

I appreciate that Improve app performance — Streamlit 0.82.0 documentation exists, but it’s an overly simplistic example that doesn’t leave people who’ve been hurt by caching in the past feeling confident they get how to do things correctly with streamlit :wink:

We have a new section on databases in our ‘latest’ version of the docs…next release, it will make it into the ‘stable’ version:

https://docs.streamlit.io/en/latest/tutorial/databases.html

If you have specific additions, we’d be happy to hear them!

Best,
Randy

Might want to consider Clickhouse in there, it’s turning out to be a very very poweful data store :wink:

Connect Streamlit to PostgreSQL — Streamlit 0.82.0 documentation is absolutely what I was looking for, many thanks!

Actualy, almost…

I have a lot of functions that look like:

def some_query(conn, p1, p2, p3):
    ...

It would be ideal if there was some way to say:

@st.cache(no_cache='conn', ttl=20)
def some_query(conn, p1, p2, p3):
    ...

Perhaps you should create a new thread, asking your specific question. We’ve gone from RDS to Clickhouse, to not caching the connection…it’s kinda hard to follow what you need :slight_smile:

Best,
Randy

Sure:

Add Clickhouse to "Connect Streamlit to databases" for requesting clickhouse docs

InternalHashError: unhashable type: 'bytearray' when using clickhouse_driver and caching · Issue #3392 · streamlit/streamlit · GitHub I’ve filed 'cos the message says to, but there’s something funky going on with streamlit trying to cache the internal state of the clickhouse_driver Client object.

I’ve also started Should st.cache() have an option to ignore certain types or parameters? to discuss the possibility of having streamlit’s caching stuff completely ignore some types of object.