""" Query history panel β€” logs every executed query with timestamp and status. Persists to ~/.dbclient/history.db (SQLite). """ import sqlite3 import datetime from pathlib import Path from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QTableWidget, QTableWidgetItem, QLineEdit, QPushButton, QHeaderView, QAbstractItemView, QMenu, QLabel, ) from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtGui import QColor, QFont _HISTORY_DB = Path.home() / ".dbclient" / "history.db" _MAX_HISTORY = 500 def _open_db() -> sqlite3.Connection: _HISTORY_DB.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(_HISTORY_DB) conn.execute(""" CREATE TABLE IF NOT EXISTS history ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, db_type TEXT, database TEXT, sql TEXT, duration REAL, status TEXT ) """) conn.commit() return conn def record_query(db_type: str, database: str, sql: str, duration: float, status: str = "OK") -> None: """Insert a history record (called from query worker result slot).""" try: conn = _open_db() conn.execute(""" INSERT INTO history (ts, db_type, database, sql, duration, status) VALUES (?, ?, ?, ?, ?, ?) """, ( datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), db_type, database, sql[:2000], round(duration, 4), status )) # Prune conn.execute(f""" DELETE FROM history WHERE id NOT IN ( SELECT id FROM history ORDER BY id DESC LIMIT {_MAX_HISTORY} ) """) conn.commit() conn.close() except Exception: pass class QueryHistoryPanel(QWidget): """Shows query history and emits signals to replay queries.""" run_query = pyqtSignal(str) # emitted when user re-runs a history item def __init__(self, parent=None): super().__init__(parent) self._build_ui() self.refresh() def _build_ui(self): root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) # Toolbar tb = QHBoxLayout() tb.setContentsMargins(6, 6, 6, 4) self._search = QLineEdit() self._search.setPlaceholderText("πŸ” Search history…") self._search.textChanged.connect(self._filter) refresh_btn = QPushButton("πŸ”„") refresh_btn.setFixedWidth(34) refresh_btn.setToolTip("Refresh") refresh_btn.clicked.connect(self.refresh) clear_btn = QPushButton("🧹") clear_btn.setFixedWidth(34) clear_btn.setToolTip("Clear all history") clear_btn.clicked.connect(self._clear_history) tb.addWidget(self._search, 1) tb.addWidget(refresh_btn) tb.addWidget(clear_btn) root.addLayout(tb) # Table self._table = QTableWidget(0, 5) self._table.setHorizontalHeaderLabels( ["Timestamp", "Database", "Duration", "Status", "SQL"]) self._table.horizontalHeader().setSectionResizeMode( QHeaderView.ResizeMode.ResizeToContents) self._table.horizontalHeader().setSectionResizeMode( 4, QHeaderView.ResizeMode.Stretch) self._table.verticalHeader().setDefaultSectionSize(24) self._table.setSelectionBehavior( QAbstractItemView.SelectionBehavior.SelectRows) self._table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) self._table.setAlternatingRowColors(True) self._table.setContextMenuPolicy( Qt.ContextMenuPolicy.CustomContextMenu) self._table.customContextMenuRequested.connect(self._context_menu) self._table.doubleClicked.connect(self._on_double_click) root.addWidget(self._table) # ── Data ───────────────────────────────────────────────────────────────── def refresh(self): self._load(self._search.text()) def _load(self, search: str = ""): try: conn = _open_db() if search: rows = conn.execute(""" SELECT ts, database, duration, status, sql FROM history WHERE sql LIKE ? ORDER BY id DESC """, (f"%{search}%",)).fetchall() else: rows = conn.execute(""" SELECT ts, database, duration, status, sql FROM history ORDER BY id DESC """).fetchall() conn.close() except Exception: rows = [] self._table.setRowCount(0) for ts, db, dur, status, sql in rows: r = self._table.rowCount() self._table.insertRow(r) items = [ ts or "", db or "", f"{dur:.3f}s" if dur else "", status or "", (sql or "").replace("\n", " ")[:200], ] for c, val in enumerate(items): item = QTableWidgetItem(val) if c == 3 and status == "ERROR": item.setForeground(QColor("#e78284")) self._table.setItem(r, c, item) def _filter(self, text: str): self._load(text) def _clear_history(self): try: conn = _open_db() conn.execute("DELETE FROM history") conn.commit() conn.close() except Exception: pass self._table.setRowCount(0) # ── Interactions ────────────────────────────────────────────────────────── def _on_double_click(self, idx): row = idx.row() sql_item = self._table.item(row, 4) if sql_item: self.run_query.emit(sql_item.text()) def _context_menu(self, pos): row = self._table.rowAt(pos.y()) if row < 0: return sql = self._table.item(row, 4) if not sql: return menu = QMenu(self) menu.addAction("β–Ά Run Query", lambda: self.run_query.emit(sql.text())) menu.addAction("πŸ“‹ Copy SQL", lambda: self._copy(sql.text())) menu.exec(self._table.viewport().mapToGlobal(pos)) @staticmethod def _copy(text: str): from PyQt6.QtWidgets import QApplication QApplication.clipboard().setText(text)