nngoandClaude Sonnet 4.6
4ccc81955e
SSL was wired through the UI and ConnectionProfile but never actually
applied in any driver or passed from the connection builder.
- main_window: pass ssl/ssl_ca/ssl_cert/ssl_key from profile into
the driver config dict so drivers can act on them
- MySQL: add ssl={ca,cert,key} to pymysql connect kwargs when enabled
- PostgreSQL: set sslmode=verify-ca (with CA) or require, plus
sslrootcert/sslcert/sslkey when provided
- MSSQL: switch Encrypt=yes + TrustServerCertificate=no when SSL is
on; Encrypt=no + TrustServerCertificate=yes otherwise
- SQLite: no change needed (local file, no network layer)
WHERE filter: add tooltip explicitly labelling the input as raw SQL
so users understand arbitrary expressions are executed directly.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
485 lines
20 KiB
Python
485 lines
20 KiB
Python
"""
|
||
Main application window.
|
||
|
||
On startup:
|
||
• All saved connection profiles are loaded from ~/.dbclient/connections.json
|
||
and shown in the sidebar in "disconnected" state.
|
||
• Double-clicking or right-click → Connect instantly re-connects.
|
||
• New Connection dialog saves the profile AND connects immediately.
|
||
• Edit / Delete work on both connected and saved-only profiles.
|
||
"""
|
||
from PyQt6.QtWidgets import (
|
||
QMainWindow, QWidget, QHBoxLayout, QVBoxLayout, QSplitter,
|
||
QTabWidget, QStatusBar, QLabel, QMessageBox, QDockWidget,
|
||
QPushButton, QApplication,
|
||
)
|
||
from PyQt6.QtCore import Qt, QTimer
|
||
from PyQt6.QtGui import QAction, QKeySequence
|
||
|
||
from app.ui.schema_browser import SchemaBrowser
|
||
from app.ui.sql_editor import SQLEditorWidget
|
||
from app.ui.table_viewer import TableViewer
|
||
from app.ui.table_structure import TableStructureView
|
||
from app.ui.query_history import QueryHistoryPanel
|
||
from app.ui.process_list import ProcessListPanel
|
||
from app.ui.import_dialog import ImportDialog
|
||
from app.ui.dump_dialog import DumpDialog
|
||
from app.ui.explain_view import ExplainPanel
|
||
from app.ui.user_manager import UserManagerPanel
|
||
from app.ui.log_viewer import LogViewer
|
||
from app.ui.connection_dialog import ConnectionDialog
|
||
from app.config.connections import load_profiles, delete_profile
|
||
from app.models.connection_model import ConnectionProfile
|
||
from app.drivers import get_driver
|
||
from app.utils.logger import get_logger
|
||
|
||
_log = get_logger(__name__)
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowTitle("DBClient")
|
||
self.resize(1400, 860)
|
||
self.setMinimumSize(1024, 640)
|
||
|
||
# profile_id → driver (only currently connected ones)
|
||
self._active_drivers: dict = {}
|
||
# profile_id → ConnectionProfile (all loaded profiles, connected or not)
|
||
self._all_profiles: dict = {}
|
||
|
||
self._build_ui()
|
||
self._build_menus()
|
||
self._build_status_bar()
|
||
|
||
# Load saved profiles after the window is shown
|
||
QTimer.singleShot(0, self._load_saved_profiles)
|
||
|
||
# ── Layout ────────────────────────────────────────────────────────────────
|
||
|
||
def _build_ui(self):
|
||
central = QWidget()
|
||
self.setCentralWidget(central)
|
||
h = QHBoxLayout(central)
|
||
h.setContentsMargins(0, 0, 0, 0)
|
||
h.setSpacing(0)
|
||
|
||
self._h_splitter = QSplitter(Qt.Orientation.Horizontal)
|
||
self._h_splitter.setHandleWidth(2)
|
||
|
||
# ── Left sidebar ──────────────────────────────────────────────────────
|
||
left = QWidget()
|
||
left.setMinimumWidth(220)
|
||
left.setMaximumWidth(420)
|
||
ll = QVBoxLayout(left)
|
||
ll.setContentsMargins(0, 0, 0, 0)
|
||
ll.setSpacing(0)
|
||
|
||
hdr = QWidget()
|
||
hdr.setObjectName("sidebarHeader")
|
||
hdr_lay = QHBoxLayout(hdr)
|
||
hdr_lay.setContentsMargins(8, 6, 8, 6)
|
||
hdr_lay.setSpacing(4)
|
||
|
||
title = QLabel(" Connections")
|
||
title.setObjectName("sidebarTitle")
|
||
f = title.font()
|
||
f.setBold(True)
|
||
title.setFont(f)
|
||
|
||
self._new_conn_btn = QPushButton("＋")
|
||
self._new_conn_btn.setObjectName("newConnBtn")
|
||
self._new_conn_btn.setFixedSize(28, 28)
|
||
self._new_conn_btn.setToolTip("New Connection (Ctrl+N)")
|
||
self._new_conn_btn.clicked.connect(self._new_connection)
|
||
|
||
hdr_lay.addWidget(title, 1)
|
||
hdr_lay.addWidget(self._new_conn_btn)
|
||
ll.addWidget(hdr)
|
||
|
||
self._schema_browser = SchemaBrowser()
|
||
# Tree → workspace wiring
|
||
self._schema_browser.open_table_viewer.connect(self._open_table_viewer)
|
||
self._schema_browser.open_table_structure.connect(self._open_table_structure)
|
||
self._schema_browser.open_sql_editor.connect(self._open_sql_editor)
|
||
self._schema_browser.run_query_requested.connect(self._paste_query)
|
||
# Saved-profile management signals
|
||
self._schema_browser.connect_requested.connect(self._connect_by_id)
|
||
self._schema_browser.edit_requested.connect(self._edit_connection)
|
||
self._schema_browser.delete_requested.connect(self._delete_connection)
|
||
ll.addWidget(self._schema_browser, 1)
|
||
|
||
# ── Right workspace ───────────────────────────────────────────────────
|
||
right = QWidget()
|
||
rl = QVBoxLayout(right)
|
||
rl.setContentsMargins(0, 0, 0, 0)
|
||
rl.setSpacing(0)
|
||
|
||
self._workspace = QTabWidget()
|
||
self._workspace.setTabsClosable(True)
|
||
self._workspace.setMovable(True)
|
||
self._workspace.tabCloseRequested.connect(self._close_tab)
|
||
self._workspace.setObjectName("workspace")
|
||
|
||
self._empty_label = QLabel(
|
||
"🔌 Double-click a saved connection to connect\n\n"
|
||
"Use ＋ to add a new connection profile."
|
||
)
|
||
self._empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self._empty_label.setObjectName("emptyLabel")
|
||
|
||
rl.addWidget(self._empty_label)
|
||
rl.addWidget(self._workspace)
|
||
self._workspace.setVisible(False)
|
||
|
||
# ── History dock ──────────────────────────────────────────────────────
|
||
self._history_panel = QueryHistoryPanel()
|
||
self._history_panel.run_query.connect(self._paste_query)
|
||
history_dock = QDockWidget("Query History", self)
|
||
history_dock.setWidget(self._history_panel)
|
||
history_dock.setMinimumHeight(120)
|
||
self.addDockWidget(Qt.DockWidgetArea.BottomDockWidgetArea, history_dock)
|
||
history_dock.setVisible(False)
|
||
self._history_dock = history_dock
|
||
|
||
self._h_splitter.addWidget(left)
|
||
self._h_splitter.addWidget(right)
|
||
self._h_splitter.setSizes([260, 1100])
|
||
h.addWidget(self._h_splitter)
|
||
|
||
# ── Menus ─────────────────────────────────────────────────────────────────
|
||
|
||
def _build_menus(self):
|
||
mb = self.menuBar()
|
||
|
||
file_menu = mb.addMenu("&File")
|
||
file_menu.addAction(self._act("＋ New Connection…", self._new_connection, "Ctrl+N"))
|
||
file_menu.addSeparator()
|
||
file_menu.addAction(self._act("Exit", QApplication.quit, "Ctrl+Q"))
|
||
|
||
view_menu = mb.addMenu("&View")
|
||
view_menu.addAction(self._act("Toggle Query History", self._toggle_history, "Ctrl+H"))
|
||
view_menu.addAction(self._act("New SQL Tab", self._new_sql_tab, "Ctrl+T"))
|
||
|
||
tools_menu = mb.addMenu("&Tools")
|
||
tools_menu.addAction(self._act("Process List…", self._open_process_list, "Ctrl+P"))
|
||
tools_menu.addSeparator()
|
||
tools_menu.addAction(self._act("Import CSV / JSON…", self._open_import_dialog))
|
||
tools_menu.addAction(self._act("Export Database Dump…", self._open_dump_dialog))
|
||
tools_menu.addSeparator()
|
||
tools_menu.addAction(self._act("User & Privilege Management…", self._open_user_manager, "Ctrl+U"))
|
||
|
||
help_menu = mb.addMenu("&Help")
|
||
help_menu.addAction(self._act("Keyboard Shortcuts", self._show_shortcuts))
|
||
help_menu.addAction(self._act("View App Logs", self._open_log_viewer, "Ctrl+L"))
|
||
help_menu.addSeparator()
|
||
help_menu.addAction(self._act("About DBClient", self._show_about))
|
||
|
||
def _act(self, text: str, slot, shortcut: str = None) -> QAction:
|
||
"""Create a QAction parented to this window (prevents GC from killing it)."""
|
||
a = QAction(text, self)
|
||
a.triggered.connect(slot)
|
||
if shortcut:
|
||
a.setShortcut(QKeySequence(shortcut))
|
||
return a
|
||
|
||
# ── Status bar ────────────────────────────────────────────────────────────
|
||
|
||
def _build_status_bar(self):
|
||
sb = QStatusBar()
|
||
self.setStatusBar(sb)
|
||
self._status_lbl = QLabel("Ready")
|
||
sb.addWidget(self._status_lbl, 1)
|
||
|
||
def _set_status(self, msg: str):
|
||
self._status_lbl.setText(msg)
|
||
|
||
# ── Startup: load saved profiles ──────────────────────────────────────────
|
||
|
||
def _load_saved_profiles(self):
|
||
"""
|
||
Load all saved connection profiles from disk and add them to the
|
||
sidebar as disconnected nodes. Called once after the window shows.
|
||
"""
|
||
profiles = load_profiles()
|
||
for profile in profiles:
|
||
self._all_profiles[profile.id] = profile
|
||
self._schema_browser.add_saved_profile(profile)
|
||
|
||
if profiles:
|
||
self._set_status(
|
||
f"Loaded {len(profiles)} saved connection(s). "
|
||
"Double-click to connect."
|
||
)
|
||
|
||
# ── Connection management ─────────────────────────────────────────────────
|
||
|
||
def _new_connection(self):
|
||
"""Open the New Connection dialog, save the profile, and connect."""
|
||
dlg = ConnectionDialog(parent=self)
|
||
if dlg.exec():
|
||
profile = dlg.profile
|
||
self._all_profiles[profile.id] = profile
|
||
# Dialog already persisted it via save_profile()
|
||
# Show it as saved first, then auto-connect
|
||
self._schema_browser.add_saved_profile(profile)
|
||
self._do_connect(profile)
|
||
|
||
def _connect_by_id(self, profile_id: str):
|
||
"""Called when user double-clicks / right-clicks Connect on a saved node."""
|
||
if self._schema_browser.is_connected(profile_id):
|
||
self._set_status("Already connected.")
|
||
return
|
||
profile = self._all_profiles.get(profile_id)
|
||
if not profile:
|
||
return
|
||
self._do_connect(profile)
|
||
|
||
def _do_connect(self, profile: ConnectionProfile):
|
||
"""Build the driver, connect, and upgrade the sidebar node."""
|
||
pid = profile.id
|
||
config = {
|
||
"host": profile.host,
|
||
"port": profile.port,
|
||
"database": profile.database,
|
||
"user": profile.username,
|
||
"password": profile.password,
|
||
"connection_timeout": profile.connection_timeout,
|
||
"ssl": profile.ssl,
|
||
"ssl_ca": profile.ssl_ca,
|
||
"ssl_cert": profile.ssl_cert,
|
||
"ssl_key": profile.ssl_key,
|
||
}
|
||
_log.info("Connecting to '%s' type=%s host=%s",
|
||
profile.name, profile.db_type, profile.host)
|
||
try:
|
||
driver = get_driver(profile.db_type, config)
|
||
driver.connect()
|
||
self._active_drivers[pid] = driver
|
||
self._schema_browser.add_connection(profile, driver)
|
||
self._show_workspace()
|
||
self._set_status(f"✅ Connected: {profile.name}")
|
||
_log.info("Connected to '%s' successfully", profile.name)
|
||
except Exception as e:
|
||
_log.error("Connection failed for '%s': %s", profile.name, e,
|
||
exc_info=True)
|
||
QMessageBox.critical(self, "Connection Error",
|
||
f"Could not connect to '{profile.name}':\n\n{e}")
|
||
|
||
def _edit_connection(self, profile_id: str):
|
||
"""Open edit dialog for a saved or connected profile."""
|
||
profile = self._all_profiles.get(profile_id)
|
||
if not profile:
|
||
return
|
||
|
||
was_connected = self._schema_browser.is_connected(profile_id)
|
||
|
||
dlg = ConnectionDialog(profile=profile, parent=self)
|
||
if not dlg.exec():
|
||
return
|
||
|
||
updated = dlg.profile
|
||
self._all_profiles[profile_id] = updated
|
||
# save_profile() was already called inside the dialog
|
||
|
||
if was_connected:
|
||
# Disconnect first, then reconnect with new credentials
|
||
self._schema_browser.remove_connection(profile_id, keep_saved=True)
|
||
self._active_drivers.pop(profile_id, None)
|
||
self._do_connect(updated)
|
||
else:
|
||
self._schema_browser.update_saved_profile(updated)
|
||
|
||
self._set_status(f"Connection '{updated.name}' updated.")
|
||
|
||
def _delete_connection(self, profile_id: str):
|
||
"""Delete a profile entirely from memory and disk."""
|
||
profile = self._all_profiles.get(profile_id)
|
||
if not profile:
|
||
return
|
||
|
||
btn = QMessageBox.warning(
|
||
self, "Delete Connection",
|
||
f"Delete the connection profile '{profile.name}'?\n\n"
|
||
"This removes it from the saved list. "
|
||
"The database itself will NOT be affected.",
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
QMessageBox.StandardButton.No,
|
||
)
|
||
if btn != QMessageBox.StandardButton.Yes:
|
||
return
|
||
|
||
# Disconnect if active
|
||
if self._schema_browser.is_connected(profile_id):
|
||
self._schema_browser.remove_connection(profile_id, keep_saved=False)
|
||
self._active_drivers.pop(profile_id, None)
|
||
else:
|
||
self._schema_browser.remove_connection(profile_id, keep_saved=False)
|
||
|
||
self._all_profiles.pop(profile_id, None)
|
||
delete_profile(profile_id)
|
||
self._set_status(f"Connection '{profile.name}' deleted.")
|
||
|
||
# ── Workspace helpers ─────────────────────────────────────────────────────
|
||
|
||
def _show_workspace(self):
|
||
self._empty_label.setVisible(False)
|
||
self._workspace.setVisible(True)
|
||
|
||
def _open_table_viewer(self, driver, database: str, table: str):
|
||
tab = TableViewer(driver, database, table)
|
||
tab.status_message.connect(self._set_status)
|
||
idx = self._workspace.addTab(tab, f"📋 {table}")
|
||
self._workspace.setCurrentIndex(idx)
|
||
|
||
def _open_table_structure(self, driver, database: str, table: str):
|
||
tab = TableStructureView(driver, database, table)
|
||
tab.status_message.connect(self._set_status)
|
||
idx = self._workspace.addTab(tab, f"🏗️ {table}")
|
||
self._workspace.setCurrentIndex(idx)
|
||
|
||
def _open_sql_editor(self, driver, database: str):
|
||
sql_widget = SQLEditorWidget()
|
||
sql_widget.status_message.connect(self._set_status)
|
||
sql_widget.new_tab(driver, database)
|
||
label = f"✏️ SQL — {database}" if database else "✏️ SQL"
|
||
idx = self._workspace.addTab(sql_widget, label)
|
||
self._workspace.setCurrentIndex(idx)
|
||
|
||
def _new_sql_tab(self):
|
||
sql_widget = SQLEditorWidget()
|
||
sql_widget.status_message.connect(self._set_status)
|
||
sql_widget.new_tab()
|
||
idx = self._workspace.addTab(sql_widget, "✏️ SQL")
|
||
self._workspace.setCurrentIndex(idx)
|
||
|
||
def _paste_query(self, sql: str):
|
||
current = self._workspace.currentWidget()
|
||
if isinstance(current, SQLEditorWidget):
|
||
tab = current.current_tab()
|
||
if tab:
|
||
tab.set_sql(sql)
|
||
return
|
||
self._new_sql_tab()
|
||
self._paste_query(sql)
|
||
|
||
def _close_tab(self, idx: int):
|
||
self._workspace.removeTab(idx)
|
||
if self._workspace.count() == 0:
|
||
self._workspace.setVisible(False)
|
||
self._empty_label.setVisible(True)
|
||
|
||
# ── Misc ──────────────────────────────────────────────────────────────────
|
||
|
||
def _toggle_history(self):
|
||
self._history_dock.setVisible(not self._history_dock.isVisible())
|
||
|
||
def _show_about(self):
|
||
QMessageBox.about(self, "About DBClient",
|
||
"<b>DBClient</b> v1.0.0<br><br>"
|
||
"Cross-platform desktop database client.<br>"
|
||
"MySQL · PostgreSQL · SQLite · SQL Server<br><br>"
|
||
"Built with Python + PyQt6.")
|
||
|
||
def _show_shortcuts(self):
|
||
QMessageBox.information(self, "Keyboard Shortcuts",
|
||
"F5 / Ctrl+Enter — Run query\n"
|
||
"Ctrl+/ — Toggle comment\n"
|
||
"Ctrl+N — New connection\n"
|
||
"Ctrl+T — New SQL tab\n"
|
||
"Ctrl+H — Toggle query history\n"
|
||
"Ctrl+U — User management\n"
|
||
"Ctrl+L — View app logs\n"
|
||
"Delete — Delete selected row (table viewer)\n"
|
||
"Ins — Add new row (table viewer)\n"
|
||
"Ctrl+Q — Quit")
|
||
|
||
def _open_log_viewer(self):
|
||
"""Open the application log viewer as a workspace tab."""
|
||
viewer = LogViewer(parent=self)
|
||
idx = self._workspace.addTab(viewer, "📋 App Logs")
|
||
self._workspace.setCurrentIndex(idx)
|
||
self._show_workspace()
|
||
|
||
def _open_process_list(self):
|
||
"""Open a Process List tab for the currently active connection."""
|
||
driver, name = self._active_driver_for_tools()
|
||
if driver is None:
|
||
return
|
||
panel = ProcessListPanel(driver, connection_name=name)
|
||
panel.status_message.connect(self._set_status)
|
||
idx = self._workspace.addTab(panel, f"⚙️ Processes — {name}")
|
||
self._workspace.setCurrentIndex(idx)
|
||
self._show_workspace()
|
||
|
||
def _open_import_dialog(self):
|
||
"""Open the Import CSV/JSON dialog targeting the active connection."""
|
||
driver, name = self._active_driver_for_tools()
|
||
if driver is None:
|
||
return
|
||
# Determine current database from the active workspace tab if possible
|
||
database = ""
|
||
current = self._workspace.currentWidget()
|
||
if hasattr(current, "_database"):
|
||
database = current._database or ""
|
||
dlg = ImportDialog(driver, database=database, table="", parent=self)
|
||
dlg.exec()
|
||
|
||
def _open_dump_dialog(self):
|
||
"""Open the Export Database Dump dialog targeting the active connection."""
|
||
driver, name = self._active_driver_for_tools()
|
||
if driver is None:
|
||
return
|
||
# Pre-select the database visible in the current workspace tab
|
||
database = ""
|
||
current = self._workspace.currentWidget()
|
||
if hasattr(current, "_database"):
|
||
database = current._database or ""
|
||
dlg = DumpDialog(driver, database=database, parent=self)
|
||
dlg.exec()
|
||
|
||
def _open_user_manager(self):
|
||
"""Open the User & Privilege Management tab."""
|
||
driver, name = self._active_driver_for_tools()
|
||
if driver is None:
|
||
return
|
||
panel = UserManagerPanel(driver, parent=self)
|
||
panel.status_message.connect(self._set_status)
|
||
idx = self._workspace.addTab(panel, f"👤 Users — {name}")
|
||
self._workspace.setCurrentIndex(idx)
|
||
self._show_workspace()
|
||
|
||
def open_explain_tab(self, driver, database: str, sql: str):
|
||
"""Open an EXPLAIN plan tab (called from SQLEditorWidget)."""
|
||
panel = ExplainPanel(driver, sql, parent=self)
|
||
panel.status_message.connect(self._set_status)
|
||
short_sql = sql[:40].replace("\n", " ") + ("…" if len(sql) > 40 else "")
|
||
idx = self._workspace.addTab(panel, f"🔎 EXPLAIN")
|
||
self._workspace.setTabToolTip(idx, short_sql)
|
||
self._workspace.setCurrentIndex(idx)
|
||
self._show_workspace()
|
||
|
||
def _active_driver_for_tools(self):
|
||
"""Return (driver, connection_name) for the first active connection,
|
||
or show a warning and return (None, '') if none are connected."""
|
||
if not self._active_drivers:
|
||
QMessageBox.information(
|
||
self, "No Active Connection",
|
||
"Connect to a database first."
|
||
)
|
||
return None, ""
|
||
pid = next(iter(self._active_drivers))
|
||
driver = self._active_drivers[pid]
|
||
profile = self._all_profiles.get(pid)
|
||
name = profile.name if profile else pid
|
||
return driver, name
|
||
|
||
def closeEvent(self, event):
|
||
for driver in self._active_drivers.values():
|
||
try:
|
||
driver.disconnect()
|
||
except Exception:
|
||
pass
|
||
event.accept()
|