Spaces:
Runtime error
Runtime error
File size: 3,311 Bytes
7b31f93 8bbdb9e 7b31f93 8bbdb9e 2288f21 0f0e2ce 8bbdb9e 0f0e2ce 7b31f93 8bbdb9e 62cc7a7 8bbdb9e d7e93db 8bbdb9e d7e93db 8bbdb9e 2288f21 8bbdb9e 7b31f93 8bbdb9e 2288f21 8bbdb9e 7b31f93 8bbdb9e 7b31f93 8bbdb9e 7b31f93 8bbdb9e 7b31f93 2288f21 8bbdb9e |
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 89 90 91 |
import asyncio
import gradio as gr
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import sessionmaker
import logging
import threading
import time
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# Global variables
db_session = None
engine = None
# Function for dynamically setting the database connection
async def set_db_connection(host, port, user, password, db_name):
global db_session, engine
try:
engine = create_async_engine(
f"mysql+aiomysql://{user}:{password}@{host}:{port}/{db_name}",
echo=False
)
Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
db_session = Session()
logger.info("Database connection established.")
return "Database connection established."
except Exception as e:
logger.error(f"Failed to establish database connection: {e}")
return f"Failed to connect to database: {e}"
# Function to update database status
def update_db_status():
global db_session
try:
if db_session:
asyncio.run(db_session.execute(select(1)))
return "Connected"
else:
return "Not connected"
except SQLAlchemyError:
return "Disconnected"
# Background task to update status
def background_update(db_status_textbox):
while True:
status = update_db_status()
db_status_textbox.value = status
logger.info(f"Database status updated: {status}")
time.sleep(60)
# Main application that runs Gradio UI and background tasks
def main():
with gr.Blocks() as demo:
gr.Markdown("# Website Monitor and Chatbot")
with gr.Row():
with gr.Column():
gr.Markdown("## Database Settings")
db_host = gr.Textbox(label="Database Host", placeholder="localhost", value="localhost")
db_port = gr.Textbox(label="Database Port", placeholder="3306", value="3306")
db_user = gr.Textbox(label="Database User", placeholder="username", value="")
db_pass = gr.Textbox(label="Database Password", placeholder="password", type="password", value="")
db_name = gr.Textbox(label="Database Name", placeholder="database_name", value="monitoring")
db_status_textbox = gr.Textbox(label="Database Status", interactive=False)
status_text = gr.Textbox(label="Status", interactive=False)
connect_button = gr.Button("Connect to Database")
# Connect button click event
connect_button.click(
set_db_connection,
inputs=[db_host, db_port, db_user, db_pass, db_name],
outputs=[status_text]
)
# Start background task to update status
threading.Thread(target=background_update, args=(db_status_textbox,), daemon=True).start()
# Launch the Gradio interface with a timeout
logger.info("Launching Gradio interface...")
demo.launch(prevent_thread_lock=True)
logger.info("Gradio interface launched successfully.")
if __name__ == "__main__":
main()
logger.info("Main function completed. App is running.") |