This is my database.py
import sqlalchemy
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy import create_engine
#from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy import engine
from sqlalchemy.engine.interfaces import CreateEnginePlugin
from sqlalchemy.orm import create_session, sessionmaker
from sqlalchemy.sql.schema import Table
from sqlalchemy.sql.sqltypes import SmallInteger, VARCHAR
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class user(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key= True)
realm = Column(String(512))
username = Column(String(512))
password = Column(String(512))
email = Column(String(512))
emailVerified = Column(SmallInteger)
verificationToken = Column(String(512))
userLanguage = Column(String(512))
def __repr__(self):
return "<testuserbot(id= '%d', realm= '%s', username='%s', password = '%s', email='%s', emailVerified= '%d', verificationToken = '%s', userLanguage='%s')>" % (
self.id, self.realm, self.username, self.password, self.email, self.emailVerified, self.verificationToken, self.userLanguage)
def UpdateUserDetails( username,email, userLanguage, password):
#engine = create_engine("mysql+pymysql://finuser:Stablehacks123@35.193.164.234:3306/finance")
engine = create_engine("mysql+pymysql://finuser@devtrustd:Stablehacks123@devtrustd.mysql.database.azure.com/finance")
metadata = MetaData()
Session = sessionmaker(bind=engine)
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
update_user = user ( username=username, email = email, userLanguage = userLanguage, password = password )
session.add(update_user)
session.commit()
action.py
from typing import Any, Text, Dict, List, Union
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
#from rasa_sdk.forms import FormAction
from rasa_sdk.events import SlotSet, EventType
from rasa_sdk.types import DomainDict
from database import UpdateUserDetails
class ValidateUserDetailsForm(Action):
def name(self) -> Text:
return "user_details_form"
def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
required_slots = ["username", "email", "userLanguage", "password"]
for slot_name in required_slots:
if tracker.slots.get(slot_name) is None:
# The slot is not yet filled, request user to fill this slot next
return [SlotSet("requested_slot", slot_name)]
# All slots filled
return [SlotSet("requested_slot", None)]
class ActionSubmit(Action):
def name(self) -> Text:
return "action_submit"
def run(
self,
dispatcher,
tracker:Tracker,
domain: "DomainDict",
) -> List[Dict[Text, Any]]:
dispatcher.utter_message(response="utter_thankyou_details", username = tracker.get_slot("username"), email = tracker.get_slot("email"), userLanguage = tracker.get_slot("userLanguage"), password = tracker.get_slot("password"))
UpdateUserDetails(tracker.get_slot("username"),tracker.get_slot("email"), tracker.get_slot("userLanguage"), tracker.get_slot("password"))
dispatcher.utter_message("Thank You for your valuable time and information. Info has been saved in our records.")
dispatcher.utter_message(template = "utter_ask_type_program_info")
return []