""" Full table data viewer — paginated grid with full CRUD support. Features: • Page-size selector: 50 / 100 / 500 / All • Add Row — dialog pre-filled with column names • Edit Row — double-click any cell to edit inline; dirty cells highlighted • Delete Row(s) — delete selected rows with confirmation • Filter (WHERE clause), Refresh • Commit / Rollback pending changes """ from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QTableView, QLabel, QPushButton, QLineEdit, QHeaderView, QAbstractItemView, QMessageBox, QDialog, QFormLayout, QDialogButtonBox, QComboBox, QScrollArea, QMenu, QApplication, ) from PyQt6.QtCore import ( Qt, QAbstractTableModel, QModelIndex, pyqtSignal, ) from PyQt6.QtGui import QColor, QBrush, QFont, QKeySequence, QShortcut from app.utils.worker import TableDataWorker # ───────────────────────────────────────────────────────────────────────────── # Editable table model # ───────────────────────────────────────────────────────────────────────────── class EditableTableModel(QAbstractTableModel): """ Tracks inline cell edits (dirty cells highlighted orange), new rows (pending insert), and rows marked for deletion. """ def __init__(self, parent=None): super().__init__(parent) self._columns: list = [] self._original: list = [] # pristine tuples self._data: list = [] # mutable lists self._dirty: set = set() # (row, col) changed cells self._pending_new: list = [] # list of dicts {col: val} self._pending_delete: list = [] # list of original tuples # ── Public API ──────────────────────────────────────────────────────────── def set_data(self, columns: list, rows: list) -> None: self.beginResetModel() self._columns = list(columns) self._original = [tuple(r) for r in rows] self._data = [list(r) for r in rows] self._dirty.clear() self._pending_new.clear() self._pending_delete.clear() self.endResetModel() def clear(self) -> None: self.set_data([], []) @property def has_changes(self) -> bool: return bool(self._dirty or self._pending_new or self._pending_delete) def column_names(self) -> list: return list(self._columns) def get_row_original(self, logical_row: int) -> tuple: return self._original[logical_row] def get_row_current(self, logical_row: int) -> list: return self._data[logical_row] def get_dirty_rows(self) -> list: """Return list of (new_data_list, original_tuple) for modified rows. Rows pending deletion are excluded — they don't need an UPDATE.""" rows_changed = set(r for r, _ in self._dirty) return [ (self._data[r], self._original[r]) for r in sorted(rows_changed) if not self._is_deleted_row(r) ] def get_pending_new(self) -> list: return list(self._pending_new) def get_pending_delete(self) -> list: return list(self._pending_delete) def discard_changes(self) -> None: for row, col in self._dirty: if row < len(self._data): self._data[row][col] = self._original[row][col] self._dirty.clear() self._pending_new.clear() self._pending_delete.clear() self.layoutChanged.emit() def mark_rows_deleted(self, logical_rows: list) -> None: """Mark rows for deletion (they stay visible but tinted red until commit).""" for r in sorted(logical_rows, reverse=True): self._pending_delete.append(self._original[r]) # Mark every cell in that row dirty-red via a special sentinel for c in range(len(self._columns)): self._dirty.add((r, c)) # will be styled via _pending_delete set self.layoutChanged.emit() def add_pending_new(self, row_dict: dict) -> None: """Queue a new row for insert.""" self._pending_new.append(row_dict) # ── QAbstractTableModel interface ───────────────────────────────────────── def rowCount(self, _parent=QModelIndex()) -> int: return len(self._data) def columnCount(self, _parent=QModelIndex()) -> int: return len(self._columns) def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole): if role == Qt.ItemDataRole.DisplayRole: if orientation == Qt.Orientation.Horizontal: return self._columns[section] if section < len(self._columns) else "" return str(section + 1) if role == Qt.ItemDataRole.FontRole and orientation == Qt.Orientation.Horizontal: f = QFont() f.setBold(True) return f return None def _is_deleted_row(self, row: int) -> bool: if row >= len(self._original): return False return self._original[row] in self._pending_delete def data(self, index: QModelIndex, role=Qt.ItemDataRole.DisplayRole): if not index.isValid(): return None r, c = index.row(), index.column() if r >= len(self._data) or c >= len(self._columns): return None val = self._data[r][c] if role == Qt.ItemDataRole.DisplayRole: return "NULL" if val is None else str(val) if role == Qt.ItemDataRole.EditRole: return "" if val is None else str(val) if role == Qt.ItemDataRole.ForegroundRole: if self._is_deleted_row(r): return QBrush(QColor("#e78284")) # red — pending delete if (r, c) in self._dirty: return QBrush(QColor("#ef9f76")) # orange — edited if val is None: return QBrush(QColor("#737994")) # grey — NULL if role == Qt.ItemDataRole.BackgroundRole: if self._is_deleted_row(r): return QBrush(QColor("#3a2530")) if (r, c) in self._dirty: return QBrush(QColor("#3a2f25")) return None def setData(self, index: QModelIndex, value, role=Qt.ItemDataRole.EditRole) -> bool: if not index.isValid() or role != Qt.ItemDataRole.EditRole: return False r, c = index.row(), index.column() if self._is_deleted_row(r): return False # don't allow editing a row queued for delete old = self._data[r][c] new = value if value.strip() != "" else None if str(old) == str(new): return False self._data[r][c] = new self._dirty.add((r, c)) self.dataChanged.emit(index, index, [role]) return True def flags(self, _index: QModelIndex): return Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable # ───────────────────────────────────────────────────────────────────────────── # Password helpers # ───────────────────────────────────────────────────────────────────────────── _PASSWORD_COL_NAMES = frozenset({ "password", "passwd", "pwd", "pass", "password_hash", "hashed_password", "passhash", "pass_hash", "user_password", "account_password", }) def _is_password_col(col_name: str) -> bool: n = col_name.lower() return n in _PASSWORD_COL_NAMES or n.endswith(("_password", "_passwd", "_pwd")) def _hash_password(plain: str) -> str: """Hash a plain-text password with bcrypt (falls back to PBKDF2).""" try: import bcrypt return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()).decode() except ImportError: import hashlib, os, base64 salt = os.urandom(16) key = hashlib.pbkdf2_hmac("sha256", plain.encode("utf-8"), salt, 260_000) return "pbkdf2:sha256:" + base64.b64encode(salt + key).decode() # ───────────────────────────────────────────────────────────────────────────── # Add / Edit row dialog # ───────────────────────────────────────────────────────────────────────────── class RowDialog(QDialog): """ Form dialog for adding or editing a single row. Password-like columns (password, passwd, pwd, …) are shown with masked input. On save the plain text is hashed automatically with bcrypt. In *edit* mode an empty password field means "keep the existing hash" — the column is omitted from the returned values dict entirely. """ def __init__(self, columns: list, initial: dict = None, title: str = "Add Row", mode: str = "add", parent=None): super().__init__(parent) self._mode = mode # "add" | "edit" self.setWindowTitle(title) self.setMinimumWidth(420) self.setModal(True) root = QVBoxLayout(self) # Scrollable form area (for tables with many columns) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setFrameShape(QScrollArea.Shape.NoFrame) form_widget = QWidget() form = QFormLayout(form_widget) form.setSpacing(8) form.setContentsMargins(8, 8, 8, 8) self._fields: dict[str, QLineEdit] = {} for col in columns: le = QLineEdit() if _is_password_col(col): le.setEchoMode(QLineEdit.EchoMode.Password) if mode == "edit": le.setPlaceholderText("Leave blank to keep current password") else: le.setPlaceholderText("Enter password") else: le.setPlaceholderText("NULL") if initial and col in initial and initial[col] is not None: le.setText(str(initial[col])) form.addRow(f"{col}:", le) self._fields[col] = le scroll.setWidget(form_widget) root.addWidget(scroll) bbox = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) bbox.accepted.connect(self.accept) bbox.rejected.connect(self.reject) root.addWidget(bbox) @property def values(self) -> dict: """Return {col: value_or_None} for all fields. Rules for password columns: • Non-empty → hash the plain text and store the hash. • Empty + add mode → store None (let the DB default apply). • Empty + edit mode → column is *omitted* (caller must not update it). """ result = {} for col, le in self._fields.items(): txt = le.text() if _is_password_col(col): if txt: result[col] = _hash_password(txt) elif self._mode == "add": result[col] = None # edit + empty → omit; caller preserves existing hash else: stripped = txt.strip() result[col] = stripped if stripped else None return result # ───────────────────────────────────────────────────────────────────────────── # Main TableViewer widget # ───────────────────────────────────────────────────────────────────────────── _PAGE_OPTIONS = [("50", 50), ("100", 100), ("All", 0)] class TableViewer(QWidget): status_message = pyqtSignal(str) def __init__(self, driver, database: str, table: str, parent=None): super().__init__(parent) self._driver = driver self._database = database self._table = table self._offset = 0 self._total = 0 self._page_size = 100 # default self._model = EditableTableModel() self._worker: TableDataWorker | None = None self._build_ui() self._load_page() # ── UI construction ─────────────────────────────────────────────────────── def _build_ui(self): root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) # ── Top toolbar ──────────────────────────────────────────────────────── tb = QHBoxLayout() tb.setContentsMargins(8, 6, 8, 4) tb.setSpacing(6) # Filter input self._filter_input = QLineEdit() self._filter_input.setPlaceholderText("WHERE … (e.g. id > 100)") self._filter_input.setFixedWidth(260) self._filter_input.setToolTip( "Raw SQL — injected directly into WHERE clause.\n" "Examples: id > 100 name LIKE 'A%' status = 'active'" ) self._filter_btn = QPushButton("🔍") self._filter_btn.setFixedWidth(34) self._filter_btn.setToolTip("Apply filter (raw SQL WHERE clause)") self._filter_btn.clicked.connect(self._apply_filter) self._filter_input.returnPressed.connect(self._apply_filter) clear_filter_btn = QPushButton("✕") clear_filter_btn.setFixedWidth(30) clear_filter_btn.setToolTip("Clear filter") clear_filter_btn.clicked.connect(self._clear_filter) self._refresh_btn = QPushButton("🔄 Refresh") self._refresh_btn.clicked.connect(self._load_page) # Separator sep = QLabel("|") sep.setStyleSheet("color:#51576d; padding:0 4px;") # CRUD buttons self._add_btn = QPushButton("➕ Add") self._edit_btn = QPushButton("✏️ Edit") self._delete_btn = QPushButton("🗑️ Delete") self._add_btn.setObjectName("crudAddBtn") self._delete_btn.setObjectName("crudDeleteBtn") self._add_btn.setFixedWidth(74) self._edit_btn.setFixedWidth(74) self._delete_btn.setFixedWidth(80) self._edit_btn.setEnabled(False) self._delete_btn.setEnabled(False) self._add_btn.clicked.connect(self._add_row) self._edit_btn.clicked.connect(self._edit_selected) self._delete_btn.clicked.connect(self._delete_selected) tb.addWidget(self._filter_input) tb.addWidget(self._filter_btn) tb.addWidget(clear_filter_btn) tb.addWidget(self._refresh_btn) tb.addWidget(sep) tb.addWidget(self._add_btn) tb.addWidget(self._edit_btn) tb.addWidget(self._delete_btn) tb.addStretch() root.addLayout(tb) # ── Table view ───────────────────────────────────────────────────────── self._table_view = QTableView() self._table_view.setModel(self._model) self._table_view.setAlternatingRowColors(True) self._table_view.setSortingEnabled(False) self._table_view.setSelectionBehavior( QAbstractItemView.SelectionBehavior.SelectRows) self._table_view.setSelectionMode( QAbstractItemView.SelectionMode.ExtendedSelection) self._table_view.horizontalHeader().setSectionResizeMode( QHeaderView.ResizeMode.Interactive) self._table_view.horizontalHeader().setStretchLastSection(True) self._table_view.verticalHeader().setDefaultSectionSize(24) self._table_view.setContextMenuPolicy( Qt.ContextMenuPolicy.CustomContextMenu) self._table_view.customContextMenuRequested.connect(self._context_menu) self._table_view.doubleClicked.connect(self._on_double_click) # Track selection to enable/disable Edit/Delete buttons. # Use both signals: clicked covers mouse, selectionChanged covers keyboard. self._table_view.clicked.connect(lambda _: self._refresh_action_states()) self._table_view.selectionModel().selectionChanged.connect( self._on_selection_changed) # Keyboard shortcuts QShortcut(QKeySequence("Delete"), self._table_view, self._delete_selected) QShortcut(QKeySequence("Ins"), self._table_view, self._add_row) root.addWidget(self._table_view, 1) # ── Pagination bar (prominent, always visible) ───────────────────────── pg_frame = QWidget() pg_frame.setObjectName("paginationBar") pg = QHBoxLayout(pg_frame) pg.setContentsMargins(10, 6, 10, 6) pg.setSpacing(4) self._first_btn = QPushButton("⏮ First") self._prev_btn = QPushButton("◀ Prev") self._next_btn = QPushButton("Next ▶") self._last_btn = QPushButton("Last ⏭") for b in (self._first_btn, self._prev_btn, self._next_btn, self._last_btn): b.setFixedWidth(80) b.setObjectName("pgBtn") self._first_btn.clicked.connect(self._first_page) self._prev_btn.clicked.connect(self._prev_page) self._next_btn.clicked.connect(self._next_page) self._last_btn.clicked.connect(self._last_page) self._page_lbl = QLabel("Page 1 / ?") self._page_lbl.setObjectName("pageLbl") self._page_lbl.setMinimumWidth(110) self._page_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter) # Page jump self._goto_input = QLineEdit() self._goto_input.setPlaceholderText("Jump to page…") self._goto_input.setFixedWidth(110) self._goto_input.returnPressed.connect(self._goto_page) goto_btn = QPushButton("Go") goto_btn.setFixedWidth(40) goto_btn.clicked.connect(self._goto_page) # Page size selector self._page_size_combo = QComboBox() for label, _ in _PAGE_OPTIONS: self._page_size_combo.addItem(label) self._page_size_combo.setCurrentIndex(1) # default = 100 self._page_size_combo.setFixedWidth(72) self._page_size_combo.setToolTip("Rows per page") self._page_size_combo.currentIndexChanged.connect(self._on_page_size_changed) self._total_lbl = QLabel("") self._total_lbl.setObjectName("rowCountLbl") pg.addWidget(self._first_btn) pg.addWidget(self._prev_btn) pg.addWidget(self._page_lbl) pg.addWidget(self._next_btn) pg.addWidget(self._last_btn) pg.addSpacing(10) pg.addWidget(self._goto_input) pg.addWidget(goto_btn) pg.addStretch() pg.addWidget(QLabel("Rows / page:")) pg.addWidget(self._page_size_combo) pg.addSpacing(8) pg.addWidget(self._total_lbl) root.addWidget(pg_frame) # ── Page size ────────────────────────────────────────────────────────────── def _on_page_size_changed(self, idx: int): self._page_size = _PAGE_OPTIONS[idx][1] self._offset = 0 self._load_page() # ── Data loading ─────────────────────────────────────────────────────────── def _load_page(self): where = self._filter_input.text().strip() limit = self._page_size if self._page_size > 0 else 999_999_999 self._worker = TableDataWorker( self._driver, self._database, self._table, where=where, limit=limit, offset=self._offset ) self._worker.finished.connect(self._on_data) self._worker.error.connect(self._on_error) self._worker.start() def _on_data(self, cols: list, rows: list, total: int): self._model.set_data(cols, rows) self._total = total self._refresh_pagination() self._refresh_action_states() self._table_view.resizeColumnsToContents() self._table_view.horizontalHeader().setStretchLastSection(True) def _on_error(self, msg: str): self._total_lbl.setText(f"Error: {msg[:80]}") self.status_message.emit(f"Error: {msg}") def _refresh_pagination(self): ps = self._page_size if self._page_size > 0 else max(self._total, 1) rows_loaded = self._model.rowCount() current_page = (self._offset // ps + 1) if ps else 1 if self._total > 0: total_pages = max(1, (self._total + ps - 1) // ps) page_str = f"Page {current_page} / {total_pages}" # showing X – Y of N row_from = self._offset + 1 row_to = min(self._offset + rows_loaded, self._total) info_str = f"Showing {row_from:,} – {row_to:,} of {self._total:,} rows" else: # total count unavailable – use row count as best guess total_pages = current_page # don't know max page_str = f"Page {current_page}" info_str = f"{rows_loaded:,} rows loaded" self._page_lbl.setText(page_str) self._total_lbl.setText(info_str) at_first = (self._offset == 0) # Next is allowed if: we have a known total and haven't reached it, # OR if a full page was returned (more rows might exist) if self._page_size > 0: if self._total > 0: at_last = (self._offset + ps) >= self._total else: at_last = rows_loaded < ps # partial page → must be last else: at_last = True # "All" mode self._first_btn.setEnabled(not at_first) self._prev_btn.setEnabled(not at_first) self._next_btn.setEnabled(not at_last) self._last_btn.setEnabled(not at_last) # ── Pagination controls ──────────────────────────────────────────────────── def _first_page(self): self._offset = 0 self._load_page() def _prev_page(self): ps = self._page_size if self._page_size > 0 else self._total self._offset = max(0, self._offset - ps) self._load_page() def _next_page(self): ps = self._page_size if self._page_size > 0 else self._total self._offset = min(self._offset + ps, max(0, self._total - ps)) self._load_page() def _last_page(self): ps = self._page_size if self._page_size > 0 else self._total self._offset = max(0, ((self._total - 1) // ps) * ps) if ps else 0 self._load_page() def _goto_page(self): ps = self._page_size if self._page_size > 0 else max(self._total, 1) try: page = int(self._goto_input.text().strip()) total_pages = max(1, (self._total + ps - 1) // ps) page = max(1, min(page, total_pages)) self._offset = (page - 1) * ps self._goto_input.clear() self._load_page() except ValueError: pass def _apply_filter(self): self._offset = 0 self._load_page() def _clear_filter(self): self._filter_input.clear() self._offset = 0 self._load_page() # ── Selection tracking ───────────────────────────────────────────────────── def _on_selection_changed(self, *_): has_sel = bool(self._table_view.selectionModel().selectedRows()) self._edit_btn.setEnabled(has_sel) self._delete_btn.setEnabled(has_sel) def _refresh_action_states(self, *_): self._on_selection_changed() def _selected_logical_rows(self) -> list: """Return sorted list of logical (source model) row indices.""" return sorted(set( idx.row() for idx in self._table_view.selectionModel().selectedRows() )) # ── Context menu ─────────────────────────────────────────────────────────── def _context_menu(self, pos): rows = self._selected_logical_rows() menu = QMenu(self) menu.addAction("➕ Add Row", self._add_row) if rows: menu.addAction("✏️ Edit Row", self._edit_selected) menu.addAction("🗑️ Delete Row", self._delete_selected) menu.addSeparator() menu.addAction("📋 Copy cell value", lambda: self._copy_cell(pos)) menu.addAction("📋 Copy row", self._copy_selected_rows) menu.exec(self._table_view.viewport().mapToGlobal(pos)) # ── Double-click: edit ──────────────────────────────────────────────────── def _on_double_click(self, index: QModelIndex): """Open edit dialog for the double-clicked row.""" row = index.row() cols = self._model.column_names() current = {cols[c]: self._model.get_row_current(row)[c] for c in range(len(cols))} dlg = RowDialog(cols, initial=current, title=f"Edit Row — {self._table}", mode="edit", parent=self) if dlg.exec(): new_vals = dlg.values original = self._model.get_row_original(row) where = {cols[i]: original[i] for i in range(len(cols))} try: self._driver.update_row( self._database, self._table, new_vals, where ) self.status_message.emit("Row updated.") self._load_page() except Exception as e: QMessageBox.critical(self, "Update Error", str(e)) # ── CRUD actions ────────────────────────────────────────────────────────── def _add_row(self): cols = self._model.column_names() if not cols: # No data loaded yet — try fetching column names try: col_infos = self._driver.get_columns(self._database, self._table) cols = [c.name for c in col_infos] except Exception as e: QMessageBox.critical(self, "Error", str(e)) return dlg = RowDialog(cols, title=f"Add Row — {self._table}", mode="add", parent=self) if dlg.exec(): row_data = {k: v for k, v in dlg.values.items()} # Remove purely empty optional fields (let DB use defaults) row_data = {k: v for k, v in row_data.items() if v is not None} try: self._driver.insert_row(self._database, self._table, row_data) self.status_message.emit("Row inserted.") self._load_page() except Exception as e: QMessageBox.critical(self, "Insert Error", str(e)) def _edit_selected(self): rows = self._selected_logical_rows() if not rows: return if len(rows) > 1: QMessageBox.information(self, "Edit Row", "Please select a single row to edit.") return self._on_double_click(self._model.index(rows[0], 0)) def _delete_selected(self): rows = self._selected_logical_rows() if not rows: return n = len(rows) btn = QMessageBox.warning( self, "Delete Row(s)", f"Are you sure you want to delete {n} row(s)?\n" "This action will be sent to the database immediately.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No, ) if btn != QMessageBox.StandardButton.Yes: return cols = self._model.column_names() errors = [] for r in rows: original = self._model.get_row_original(r) where = {cols[i]: original[i] for i in range(len(cols))} try: self._driver.delete_row(self._database, self._table, where) except Exception as e: errors.append(str(e)) if errors: QMessageBox.critical(self, "Delete Error", "\n".join(errors)) else: self.status_message.emit(f"{n} row(s) deleted.") self._load_page() # ── Copy helpers ────────────────────────────────────────────────────────── def _copy_cell(self, pos): idx = self._table_view.indexAt(pos) if idx.isValid(): val = self._model.data(idx, Qt.ItemDataRole.DisplayRole) or "" QApplication.clipboard().setText(str(val)) def _copy_selected_rows(self): rows = self._selected_logical_rows() cols = self._model.column_names() lines = ["\t".join(cols)] for r in rows: row_data = self._model.get_row_current(r) lines.append("\t".join("" if v is None else str(v) for v in row_data)) QApplication.clipboard().setText("\n".join(lines))