How do I switch the object of my database "MySQL" between two actions?

Question

I am creating my own chatbot that can interface with my internal database, and I want to avoid creating a database connection every time I call an action, instead I would like to pass the MySQL connection object as a parameter. `class ConnectToDatabase(Action):

def name(self):
    return "action_connect_to_database"

def run(self, dispatcher, tracker, domain):
    try:
        conn = mysql.connector.connect(
           "Parameters of my db"
        )
        conn_json = json.dumps(conn, default=str)
        return [SlotSet("database_connection", conn_json)]
    except Exception as e:
         dispatcher.utter_message(text=f"Error: {str(e)}")

class ExecuteDatabaseQuery(Action):

def name(self):
    return "action_execute_database_query"

def run(self, dispatcher, tracker, domain):
    try:
        conn = (tracker.get_slot("database_connection"))
        conn = json.loads(conn)
        print("Conn3:", str(type(conn)))
        ...`

But this gives me an error, specifically in the ExecuteDatabaseQuery function the "conn" object turns out to be empty, so the way "database_connection" is passed into the slot is wrong. Can anyone tell me why this is the case?

I have been trying to pass an object, specifically the database connection, between two python actions in the RASA Framework with the goal of optimizing resources.

Answer 1

The reason why the "conn" object is empty in the ExecuteDatabaseQuery function is because the json.loads() function is trying to load a string, but the "database_connection" slot is actually a JSON object.

To fix this, you need to change the way that you are serializing and deserializing the database connection object. Instead of using json.dumps() and json.loads(), you can use the rasa.shared.utils.io.json_to_dict() and rasa.shared.utils.io.dict_to_json() functions.

Here is the updated code for the ConnectToDatabase action:

class ConnectToDatabase(Action):
    def name(self):
        return "action_connect_to_database"

    def run(self, dispatcher, tracker, domain):
        try:
            conn = mysql.connector.connect(
                "Parameters of my db"
            )
            return [SlotSet("database_connection", rasa.shared.utils.io.dict_to_json(conn))]
        except Exception as e:
            dispatcher.utter_message(text=f"Error: {str(e)}")

And here is the updated code for the ExecuteDatabaseQuery action:

class ExecuteDatabaseQuery(Action):
    def name(self):
        return "action_execute_database_query"

    def run(self, dispatcher, tracker, domain):
        try:
            conn = rasa.shared.utils.io.json_to_dict(tracker.get_slot("database_connection"))
            print("Conn3:", str(type(conn)))
            ...
        except Exception as e:
            dispatcher.utter_message(text=f"Error: {str(e)}")

With this change, the "conn" object will now be properly serialized and deserialized, and it will not be empty in the ExecuteDatabaseQuery function.