Spaces:
Running
Running
import pymysql | |
import json | |
import pandas as pd | |
import re | |
import streamlit as st | |
# Database connection | |
def initialize_database(): | |
# Database Connection | |
db_params = {"host": st.secrets["host"], | |
"user": st.secrets["user"], | |
"password": st.secrets["password"], | |
"port": st.secrets["port"], | |
"database": st.secrets["database"] | |
} | |
db = pymysql.connect(**db_params) | |
return db | |
def execute_query(query): | |
db = initialize_database() | |
cursor = db.cursor() | |
try: | |
cursor.execute(query) | |
description = cursor.description | |
result = cursor.fetchall() # Fetch all rows from the result set | |
db.commit() | |
return description, result | |
except Exception as e: | |
print("Error executing query:", e) | |
db.rollback() | |
return None # Return None if an error occurs | |
finally: | |
db.close() | |
def _get_details_mantra_json(self, query): | |
description, data = execute_query(query) | |
df = pd.DataFrame(data) | |
df.columns = [x[0] for x in description] | |
mantra_json = df['mantra_json'].values[0] | |
cleaned_data = re.sub('<[^<]+?>', '', mantra_json) | |
return json.loads(cleaned_data) | |