Spaces:
Running
Running
File size: 2,489 Bytes
9a19c9e b7c24ab 9a19c9e ca1820c 9a19c9e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
import pymysql
import streamlit as st
import logging
import json
import pandas as pd
import re
import os
from langchain_community.utilities.sql_database import SQLDatabase
db_params = {"host": os.getenv("DB_HOST"),
"user": os.getenv("DB_USER"),
"password": os.getenv("DB_PASSWORD"),
"port": 3306,
"database":os.getenv("DB")
}
def initialize_database():
try:
# Database Connection
db = pymysql.connect(**db_params)
st.success("Database connection successful!")
return db
except Exception as e:
st.error(f"Database connection failed: {e}")
return None
def get_db():
try:
db = SQLDatabase.from_uri(
f"mysql+pymysql://{db_params['user']}:{db_params['password']}@{db_params['host']}/{db_params['database']}",
include_tables=['term_details_modified', 'veda_content_details', 'veda_content_modified']
)
#st.success("Database connection successful!")
return db
except Exception as e:
st.error(f"Database connection failed: {e}")
return None
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 execute_sql_query(query, parameters=None):
# Establish database connection and execute SQL query
db = initialize_database()
cursor = db.cursor(pymysql.cursors.DictCursor) # Use dictionary cursor to retrieve data as dictionaries
try:
if parameters:
cursor.execute(query, parameters)
else:
cursor.execute(query)
results = cursor.fetchall()
return results
except Exception as e:
logging.error(f"Error executing SQL query: {e}")
return None
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)
|