from dash import Output, Input, html
import dash_bootstrap_components as dbc
from utils.logger import get_logger
from database.connection import DatabaseManager
from datetime import datetime, timedelta
from sqlalchemy import text
from config.constants.system_health_constants import (
DATABASE_RECENT_ACTIVITY_HOURS,
DATABASE_LARGEST_TABLES_LIMIT
)
from database.operations import get_database_operations
logger = get_logger("default_logger")
def register_database_callbacks(app):
"""Register database status and statistics callbacks."""
# Database Status and Statistics
@app.callback(
[Output('database-status', 'children'),
Output('database-stats', 'children')],
Input('interval-component', 'n_intervals')
)
def update_database_status(n_intervals):
"""Update database connection status and statistics."""
try:
db_status = _get_database_status()
db_stats = _get_database_statistics()
return db_status, db_stats
except Exception as e:
logger.error(f"Error updating database status: {e}")
error_alert = dbc.Alert(
f"Error: {str(e)}",
color="danger",
dismissable=True
)
return error_alert, error_alert
def _get_database_status() -> html.Div:
"""Get detailed database status."""
db_operations = get_database_operations(logger)
try:
is_connected = db_operations.health_check()
current_time = datetime.now().strftime('%H:%M:%S')
if is_connected:
status_badge = dbc.Badge("Database Connected", color="success")
# Placeholder for version and connections, as get_stats will provide more detailed info
details_text = html.P("Details available in Database Statistics section.", className="mb-0")
else:
status_badge = dbc.Badge("Database Disconnected", color="danger")
details_text = html.P("Could not connect to the database.", className="mb-0")
return html.Div([
dbc.Row([
dbc.Col(status_badge, width="auto"),
dbc.Col(html.P(f"Checked: {current_time}", className="text-muted"), width="auto")
], align="center", className="mb-2"),
details_text
])
except Exception as e:
logger.error(f"Error connecting to database: {e}")
return dbc.Alert(f"Error connecting to database: {e}", color="danger")
def _get_database_statistics() -> html.Div:
"""Get database statistics."""
db_operations = get_database_operations(logger)
try:
stats = db_operations.get_stats()
if not stats.get('healthy'):
return dbc.Alert(f"Database statistics unavailable: {stats.get('error', 'Connection failed')}", color="warning")
components = [
dbc.Row([
dbc.Col(html.Strong(f"Bots:")),
dbc.Col(f"{stats.get('bot_count', 'N/A')}", className="text-end")
]),
dbc.Row([
dbc.Col(html.Strong(f"Candles:")),
dbc.Col(f"{stats.get('candle_count', 'N/A')}", className="text-end")
]),
dbc.Row([
dbc.Col(html.Strong(f"Raw Trades:")),
dbc.Col(f"{stats.get('raw_trade_count', 'N/A')}", className="text-end")
]),
# TODO: Integrate detailed table stats, recent activity from `database.operations` if available
# Currently, `get_stats` does not provide this granular data directly.
]
return html.Div(components)
except Exception as e:
logger.error(f"Error loading database stats: {e}")
return dbc.Alert(f"Error loading database stats: {e}", color="danger")