Spaces:
Running
Running
| import pymysql | |
| import json | |
| import pandas as pd | |
| import re | |
| import streamlit as st | |
| import logging | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) # Set logging level to INFO | |
| # Database connection | |
| def initialize_database(): | |
| try: | |
| # Database Connection | |
| db_params = {"host": st.secrets["host"], | |
| "user": st.secrets["username"], | |
| "password": st.secrets["password"], | |
| "port": int(st.secrets["port"]), | |
| "database": st.secrets["database"], | |
| } | |
| db = pymysql.connect(**db_params) | |
| logging.info("Connected to the database successfully!") | |
| return db | |
| except pymysql.MySQLError as e: | |
| logging.error("Error connecting to the database: %s", e) | |
| raise # Re-raise the exception to propagate it up the call stack | |
| def execute_query(query): | |
| print(f"Db initilaizing....") | |
| db = initialize_database() | |
| print(f"Db initialized....") | |
| cursor = db.cursor() | |
| try: | |
| cursor.execute(query) | |
| description = cursor.description | |
| result = cursor.fetchall() # Fetch all rows from the result set | |
| db.commit() | |
| logging.info("Query executed successfully: %s", query) | |
| return description, result | |
| except Exception as e: | |
| logging.error("Error executing query: %s", e) | |
| db.rollback() | |
| return None # Return None if an error occurs | |
| finally: | |
| db.close() | |
| def get_details_mantra_json(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) | |