632 lines
23 KiB
Python
632 lines
23 KiB
Python
"""
|
||
User & Privilege Management panel.
|
||
|
||
Supports MySQL and PostgreSQL.
|
||
SQLite and MSSQL show an informational message (no user management via this UI).
|
||
|
||
Features
|
||
--------
|
||
• List all database users with host (MySQL) or connection limit (PostgreSQL)
|
||
• Create user (username + password + optional host for MySQL)
|
||
• Drop user with confirmation
|
||
• View per-database GRANT privileges for the selected user
|
||
• GRANT / REVOKE individual privileges on a database
|
||
|
||
All DDL operations execute synchronously in a background QThread to keep
|
||
the UI responsive.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from PyQt6.QtWidgets import (
|
||
QWidget, QVBoxLayout, QHBoxLayout, QSplitter,
|
||
QTableWidget, QTableWidgetItem, QHeaderView,
|
||
QLabel, QPushButton, QComboBox, QCheckBox,
|
||
QDialog, QFormLayout, QLineEdit, QDialogButtonBox,
|
||
QMessageBox, QGroupBox, QAbstractItemView,
|
||
)
|
||
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
||
from PyQt6.QtGui import QColor, QFont
|
||
|
||
|
||
# ── Standard SQL privileges offered in the GRANT UI ──────────────────────────
|
||
|
||
_MYSQL_PRIVS = [
|
||
"SELECT", "INSERT", "UPDATE", "DELETE",
|
||
"CREATE", "DROP", "ALTER", "INDEX",
|
||
"CREATE VIEW", "SHOW VIEW",
|
||
"CREATE ROUTINE", "ALTER ROUTINE", "EXECUTE",
|
||
"REFERENCES", "TRIGGER", "LOCK TABLES",
|
||
"CREATE TEMPORARY TABLES",
|
||
]
|
||
|
||
_PG_PRIVS = [
|
||
"SELECT", "INSERT", "UPDATE", "DELETE",
|
||
"TRUNCATE", "REFERENCES", "TRIGGER",
|
||
"CREATE", "CONNECT", "TEMPORARY",
|
||
"EXECUTE", "USAGE",
|
||
]
|
||
|
||
|
||
# ── Background worker ─────────────────────────────────────────────────────────
|
||
|
||
class _UserWorker(QThread):
|
||
"""Run an arbitrary callable in a background thread."""
|
||
result = pyqtSignal(object)
|
||
error = pyqtSignal(str)
|
||
|
||
def __init__(self, fn, *args, parent=None):
|
||
super().__init__(parent)
|
||
self._fn = fn
|
||
self._args = args
|
||
|
||
def run(self):
|
||
try:
|
||
self.result.emit(self._fn(*self._args))
|
||
except Exception as e:
|
||
self.error.emit(str(e))
|
||
|
||
|
||
# ── Driver-level helpers (attached at runtime, no driver subclassing needed) ──
|
||
|
||
def _mysql_get_users(driver) -> list[dict]:
|
||
"""Return list of {user, host, super} from MySQL.user table."""
|
||
cols, rows, _ = driver.execute_query(
|
||
"SELECT User, Host, Super_priv FROM mysql.user ORDER BY User, Host"
|
||
)
|
||
return [{"user": r[0], "host": r[1], "super": r[2]} for r in rows]
|
||
|
||
|
||
def _mysql_get_grants(driver, user: str, host: str) -> list[str]:
|
||
cols, rows, _ = driver.execute_query(
|
||
f"SHOW GRANTS FOR %s@%s", (user, host) # type: ignore[arg-type]
|
||
)
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
def _mysql_create_user(driver, user: str, host: str, password: str):
|
||
driver.execute_query(
|
||
"CREATE USER %s@%s IDENTIFIED BY %s", (user, host, password)
|
||
)
|
||
|
||
|
||
def _mysql_drop_user(driver, user: str, host: str):
|
||
driver.execute_query(f"DROP USER %s@%s", (user, host))
|
||
|
||
|
||
def _mysql_grant(driver, privs: list[str], database: str,
|
||
user: str, host: str):
|
||
priv_str = ", ".join(privs) if privs else "USAGE"
|
||
driver.execute_query(
|
||
f"GRANT {priv_str} ON `{database}`.* TO %s@%s", (user, host)
|
||
)
|
||
driver.execute_query("FLUSH PRIVILEGES")
|
||
|
||
|
||
def _mysql_revoke(driver, privs: list[str], database: str,
|
||
user: str, host: str):
|
||
priv_str = ", ".join(privs) if privs else "USAGE"
|
||
driver.execute_query(
|
||
f"REVOKE {priv_str} ON `{database}`.* FROM %s@%s", (user, host)
|
||
)
|
||
driver.execute_query("FLUSH PRIVILEGES")
|
||
|
||
|
||
# ── PostgreSQL helpers ────────────────────────────────────────────────────────
|
||
|
||
def _pg_get_users(driver) -> list[dict]:
|
||
cols, rows, _ = driver.execute_query(
|
||
"SELECT usename, usesuper, usecreatedb FROM pg_user ORDER BY usename"
|
||
)
|
||
return [{"user": r[0], "host": "", "super": "Y" if r[1] else "N",
|
||
"createdb": "Y" if r[2] else "N"} for r in rows]
|
||
|
||
|
||
def _pg_get_grants(driver, user: str, _host: str) -> list[str]:
|
||
cols, rows, _ = driver.execute_query("""
|
||
SELECT grantee, table_catalog, table_schema, table_name,
|
||
string_agg(privilege_type, ', ') AS privileges
|
||
FROM information_schema.role_table_grants
|
||
WHERE grantee = %s
|
||
GROUP BY grantee, table_catalog, table_schema, table_name
|
||
ORDER BY table_catalog, table_schema, table_name
|
||
""", (user,))
|
||
return [
|
||
f"GRANT {r[4]} ON {r[1]}.{r[2]}.{r[3]} TO {r[0]}"
|
||
for r in rows
|
||
]
|
||
|
||
|
||
def _pg_create_user(driver, user: str, _host: str, password: str):
|
||
driver.execute_query(
|
||
f"CREATE USER \"{user}\" WITH PASSWORD %s", (password,)
|
||
)
|
||
|
||
|
||
def _pg_drop_user(driver, user: str, _host: str):
|
||
driver.execute_query(f'DROP USER "{user}"')
|
||
|
||
|
||
def _pg_grant(driver, privs: list[str], database: str, user: str, _host: str):
|
||
priv_str = ", ".join(privs) if privs else "CONNECT"
|
||
driver.execute_query(
|
||
f'GRANT {priv_str} ON DATABASE "{database}" TO "{user}"'
|
||
)
|
||
|
||
|
||
def _pg_revoke(driver, privs: list[str], database: str, user: str, _host: str):
|
||
priv_str = ", ".join(privs) if privs else "CONNECT"
|
||
driver.execute_query(
|
||
f'REVOKE {priv_str} ON DATABASE "{database}" FROM "{user}"'
|
||
)
|
||
|
||
|
||
# ── Dialogs ───────────────────────────────────────────────────────────────────
|
||
|
||
class _CreateUserDialog(QDialog):
|
||
def __init__(self, db_type: str, parent=None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("Create User")
|
||
self.setModal(True)
|
||
self.setMinimumWidth(360)
|
||
self._db_type = db_type
|
||
|
||
root = QVBoxLayout(self)
|
||
form = QFormLayout()
|
||
form.setSpacing(8)
|
||
|
||
self._user_edit = QLineEdit()
|
||
self._user_edit.setPlaceholderText("e.g. app_user")
|
||
form.addRow("Username:", self._user_edit)
|
||
|
||
if db_type == "mysql":
|
||
self._host_edit = QLineEdit()
|
||
self._host_edit.setText("%")
|
||
self._host_edit.setPlaceholderText("% = any host")
|
||
form.addRow("Host:", self._host_edit)
|
||
|
||
self._pass_edit = QLineEdit()
|
||
self._pass_edit.setEchoMode(QLineEdit.EchoMode.Password)
|
||
self._pass_edit.setPlaceholderText("Password")
|
||
form.addRow("Password:", self._pass_edit)
|
||
|
||
self._pass2_edit = QLineEdit()
|
||
self._pass2_edit.setEchoMode(QLineEdit.EchoMode.Password)
|
||
self._pass2_edit.setPlaceholderText("Confirm password")
|
||
form.addRow("Confirm:", self._pass2_edit)
|
||
|
||
root.addLayout(form)
|
||
|
||
bbox = QDialogButtonBox(
|
||
QDialogButtonBox.StandardButton.Ok |
|
||
QDialogButtonBox.StandardButton.Cancel,
|
||
)
|
||
bbox.accepted.connect(self._validate)
|
||
bbox.rejected.connect(self.reject)
|
||
root.addWidget(bbox)
|
||
|
||
def _validate(self):
|
||
if not self._user_edit.text().strip():
|
||
QMessageBox.warning(self, "Validation", "Username is required.")
|
||
return
|
||
if self._pass_edit.text() != self._pass2_edit.text():
|
||
QMessageBox.warning(self, "Validation", "Passwords do not match.")
|
||
return
|
||
self.accept()
|
||
|
||
@property
|
||
def username(self) -> str:
|
||
return self._user_edit.text().strip()
|
||
|
||
@property
|
||
def host(self) -> str:
|
||
if self._db_type == "mysql":
|
||
return self._host_edit.text().strip() or "%"
|
||
return ""
|
||
|
||
@property
|
||
def password(self) -> str:
|
||
return self._pass_edit.text()
|
||
|
||
|
||
class _GrantDialog(QDialog):
|
||
"""Select privileges + target database and grant/revoke."""
|
||
|
||
def __init__(self, driver, db_type: str, databases: list[str],
|
||
user: str, host: str, mode: str = "grant", parent=None):
|
||
super().__init__(parent)
|
||
self._driver = driver
|
||
self._db_type = db_type
|
||
self._user = user
|
||
self._host = host
|
||
self._mode = mode # "grant" | "revoke"
|
||
self.setWindowTitle(f"{'Grant to' if mode == 'grant' else 'Revoke from'} {user}")
|
||
self.setModal(True)
|
||
self.setMinimumSize(420, 480)
|
||
self._build_ui(databases)
|
||
|
||
def _build_ui(self, databases: list[str]):
|
||
root = QVBoxLayout(self)
|
||
|
||
form = QFormLayout()
|
||
self._db_combo = QComboBox()
|
||
for db in databases:
|
||
self._db_combo.addItem(db)
|
||
form.addRow("Database:", self._db_combo)
|
||
root.addLayout(form)
|
||
|
||
# Privilege checkboxes
|
||
priv_box = QGroupBox("Privileges")
|
||
pb_lay = QVBoxLayout(priv_box)
|
||
privs = _MYSQL_PRIVS if self._db_type == "mysql" else _PG_PRIVS
|
||
|
||
self._priv_checks: list[QCheckBox] = []
|
||
for priv in privs:
|
||
cb = QCheckBox(priv)
|
||
pb_lay.addWidget(cb)
|
||
self._priv_checks.append(cb)
|
||
|
||
sel_row = QHBoxLayout()
|
||
sel_all = QPushButton("All")
|
||
sel_all.setFixedWidth(50)
|
||
sel_all.clicked.connect(lambda: [cb.setChecked(True)
|
||
for cb in self._priv_checks])
|
||
sel_none = QPushButton("None")
|
||
sel_none.setFixedWidth(50)
|
||
sel_none.clicked.connect(lambda: [cb.setChecked(False)
|
||
for cb in self._priv_checks])
|
||
sel_row.addWidget(sel_all)
|
||
sel_row.addWidget(sel_none)
|
||
sel_row.addStretch()
|
||
pb_lay.insertLayout(0, sel_row)
|
||
|
||
root.addWidget(priv_box, 1)
|
||
|
||
verb = "Grant" if self._mode == "grant" else "Revoke"
|
||
bbox = QDialogButtonBox()
|
||
ok = bbox.addButton(verb, QDialogButtonBox.ButtonRole.AcceptRole)
|
||
ok.clicked.connect(self._execute)
|
||
bbox.addButton(QDialogButtonBox.StandardButton.Cancel).clicked.connect(
|
||
self.reject)
|
||
root.addWidget(bbox)
|
||
|
||
def _execute(self):
|
||
privs = [cb.text() for cb in self._priv_checks if cb.isChecked()]
|
||
if not privs:
|
||
QMessageBox.warning(self, "No Privilege",
|
||
"Select at least one privilege.")
|
||
return
|
||
db = self._db_combo.currentText()
|
||
|
||
try:
|
||
if self._db_type == "mysql":
|
||
fn = _mysql_grant if self._mode == "grant" else _mysql_revoke
|
||
else:
|
||
fn = _pg_grant if self._mode == "grant" else _pg_revoke
|
||
fn(self._driver, privs, db, self._user, self._host)
|
||
self.accept()
|
||
except Exception as e:
|
||
QMessageBox.critical(self, "Error", str(e))
|
||
|
||
@property
|
||
def selected_privs(self) -> list[str]:
|
||
return [cb.text() for cb in self._priv_checks if cb.isChecked()]
|
||
|
||
|
||
# ── Main Panel ────────────────────────────────────────────────────────────────
|
||
|
||
class UserManagerPanel(QWidget):
|
||
"""
|
||
User & privilege management workspace tab.
|
||
|
||
Parameters
|
||
----------
|
||
driver : BaseDriver — must already be connected
|
||
parent : QWidget
|
||
"""
|
||
status_message = pyqtSignal(str)
|
||
|
||
def __init__(self, driver, parent=None):
|
||
super().__init__(parent)
|
||
self._driver = driver
|
||
self._db_type = getattr(driver, "db_type", "").lower()
|
||
self._databases: list[str] = []
|
||
self._current_user: str = ""
|
||
self._current_host: str = ""
|
||
self._worker: _UserWorker | None = None
|
||
self._build_ui()
|
||
self._load_users()
|
||
self._load_databases()
|
||
|
||
# ── UI construction ───────────────────────────────────────────────────────
|
||
|
||
def _build_ui(self):
|
||
root = QVBoxLayout(self)
|
||
root.setContentsMargins(0, 0, 0, 0)
|
||
root.setSpacing(0)
|
||
|
||
# ── Header bar ────────────────────────────────────────────────────────
|
||
hdr = QHBoxLayout()
|
||
hdr.setContentsMargins(8, 6, 8, 6)
|
||
hdr.setSpacing(6)
|
||
|
||
title = QLabel("👤 User & Privilege Management")
|
||
f = title.font()
|
||
f.setBold(True)
|
||
f.setPointSize(11)
|
||
title.setFont(f)
|
||
title.setObjectName("structureTitle")
|
||
|
||
self._refresh_btn = QPushButton("🔄 Refresh")
|
||
self._refresh_btn.setFixedWidth(90)
|
||
self._refresh_btn.clicked.connect(self._load_users)
|
||
|
||
hdr.addWidget(title)
|
||
hdr.addStretch()
|
||
hdr.addWidget(self._refresh_btn)
|
||
root.addLayout(hdr)
|
||
|
||
# ── Unsupported banner (hidden for MySQL/PG) ──────────────────────────
|
||
self._unsupported_lbl = QLabel(
|
||
"ℹ️ User management is not available for this database type.\n"
|
||
" (Supports MySQL and PostgreSQL only.)"
|
||
)
|
||
self._unsupported_lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self._unsupported_lbl.setObjectName("emptyLabel")
|
||
root.addWidget(self._unsupported_lbl)
|
||
|
||
if self._db_type not in ("mysql", "postgresql"):
|
||
self._unsupported_lbl.setVisible(True)
|
||
return
|
||
self._unsupported_lbl.setVisible(False)
|
||
|
||
# ── Main splitter: user list (left) / grants detail (right) ──────────
|
||
splitter = QSplitter(Qt.Orientation.Horizontal)
|
||
|
||
# ── Left: user table ──────────────────────────────────────────────────
|
||
left = QWidget()
|
||
ll = QVBoxLayout(left)
|
||
ll.setContentsMargins(0, 0, 0, 0)
|
||
|
||
ll.addWidget(self._make_section_label("Users"))
|
||
|
||
# Toolbar
|
||
tb = QHBoxLayout()
|
||
tb.setContentsMargins(4, 0, 4, 4)
|
||
self._add_user_btn = QPushButton("➕ Create User")
|
||
self._drop_user_btn = QPushButton("🗑️ Drop User")
|
||
self._drop_user_btn.setObjectName("crudDeleteBtn")
|
||
self._drop_user_btn.setEnabled(False)
|
||
self._add_user_btn.clicked.connect(self._create_user)
|
||
self._drop_user_btn.clicked.connect(self._drop_user)
|
||
tb.addWidget(self._add_user_btn)
|
||
tb.addWidget(self._drop_user_btn)
|
||
tb.addStretch()
|
||
ll.addLayout(tb)
|
||
|
||
self._user_table = QTableWidget()
|
||
self._user_table.setAlternatingRowColors(True)
|
||
self._user_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
|
||
self._user_table.setSelectionBehavior(
|
||
QAbstractItemView.SelectionBehavior.SelectRows)
|
||
self._user_table.setSelectionMode(
|
||
QAbstractItemView.SelectionMode.SingleSelection)
|
||
self._user_table.verticalHeader().setDefaultSectionSize(24)
|
||
self._user_table.itemSelectionChanged.connect(self._on_user_selected)
|
||
ll.addWidget(self._user_table)
|
||
|
||
splitter.addWidget(left)
|
||
|
||
# ── Right: grants panel ───────────────────────────────────────────────
|
||
right = QWidget()
|
||
rl = QVBoxLayout(right)
|
||
rl.setContentsMargins(0, 0, 0, 0)
|
||
|
||
rl.addWidget(self._make_section_label("Grants for selected user"))
|
||
|
||
# Grant toolbar
|
||
gtb = QHBoxLayout()
|
||
gtb.setContentsMargins(4, 0, 4, 4)
|
||
self._grant_btn = QPushButton("＋ Grant")
|
||
self._revoke_btn = QPushButton("－ Revoke")
|
||
self._grant_btn.setEnabled(False)
|
||
self._revoke_btn.setEnabled(False)
|
||
self._grant_btn.clicked.connect(self._grant_privs)
|
||
self._revoke_btn.clicked.connect(self._revoke_privs)
|
||
gtb.addWidget(self._grant_btn)
|
||
gtb.addWidget(self._revoke_btn)
|
||
gtb.addStretch()
|
||
rl.addLayout(gtb)
|
||
|
||
self._grants_table = QTableWidget(0, 1)
|
||
self._grants_table.setHorizontalHeaderLabels(["GRANT Statement"])
|
||
self._grants_table.horizontalHeader().setSectionResizeMode(
|
||
QHeaderView.ResizeMode.Stretch)
|
||
self._grants_table.setAlternatingRowColors(True)
|
||
self._grants_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
|
||
self._grants_table.verticalHeader().setDefaultSectionSize(24)
|
||
rl.addWidget(self._grants_table)
|
||
|
||
splitter.addWidget(right)
|
||
splitter.setSizes([340, 560])
|
||
|
||
root.addWidget(splitter, 1)
|
||
|
||
# ── Status bar ────────────────────────────────────────────────────────
|
||
self._status_lbl = QLabel("")
|
||
self._status_lbl.setObjectName("rowCountLbl")
|
||
self._status_lbl.setContentsMargins(8, 4, 0, 4)
|
||
root.addWidget(self._status_lbl)
|
||
|
||
@staticmethod
|
||
def _make_section_label(text: str) -> QLabel:
|
||
lbl = QLabel(f" {text}")
|
||
lbl.setObjectName("sidebarTitle")
|
||
lbl.setContentsMargins(0, 4, 0, 4)
|
||
f = lbl.font()
|
||
f.setBold(True)
|
||
lbl.setFont(f)
|
||
return lbl
|
||
|
||
# ── Data loading ──────────────────────────────────────────────────────────
|
||
|
||
def _load_users(self):
|
||
if self._db_type not in ("mysql", "postgresql"):
|
||
return
|
||
fn = _mysql_get_users if self._db_type == "mysql" else _pg_get_users
|
||
self._worker = _UserWorker(fn, self._driver, parent=self)
|
||
self._worker.result.connect(self._populate_users)
|
||
self._worker.error.connect(
|
||
lambda e: self._set_status(f"Error loading users: {e}"))
|
||
self._worker.start()
|
||
|
||
def _load_databases(self):
|
||
try:
|
||
self._databases = self._driver.get_databases()
|
||
except Exception:
|
||
self._databases = []
|
||
|
||
def _populate_users(self, users: list[dict]):
|
||
t = self._user_table
|
||
t.clear()
|
||
|
||
if self._db_type == "mysql":
|
||
t.setColumnCount(3)
|
||
t.setHorizontalHeaderLabels(["Username", "Host", "Super"])
|
||
else:
|
||
t.setColumnCount(3)
|
||
t.setHorizontalHeaderLabels(["Username", "Superuser", "CreateDB"])
|
||
|
||
t.setRowCount(len(users))
|
||
t.horizontalHeader().setSectionResizeMode(
|
||
QHeaderView.ResizeMode.ResizeToContents)
|
||
t.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
|
||
|
||
for r, u in enumerate(users):
|
||
if self._db_type == "mysql":
|
||
vals = [u["user"], u["host"], u.get("super", "")]
|
||
else:
|
||
vals = [u["user"], u.get("super", ""), u.get("createdb", "")]
|
||
for c, v in enumerate(vals):
|
||
item = QTableWidgetItem(str(v))
|
||
item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
t.setItem(r, c, item)
|
||
|
||
self._set_status(f"{len(users)} user(s)")
|
||
|
||
def _on_user_selected(self):
|
||
rows = self._user_table.selectionModel().selectedRows()
|
||
if not rows:
|
||
self._drop_user_btn.setEnabled(False)
|
||
self._grant_btn.setEnabled(False)
|
||
self._revoke_btn.setEnabled(False)
|
||
return
|
||
|
||
r = rows[0].row()
|
||
self._current_user = self._user_table.item(r, 0).text()
|
||
self._current_host = (
|
||
self._user_table.item(r, 1).text()
|
||
if self._db_type == "mysql"
|
||
else ""
|
||
)
|
||
|
||
self._drop_user_btn.setEnabled(True)
|
||
self._grant_btn.setEnabled(True)
|
||
self._revoke_btn.setEnabled(True)
|
||
self._load_grants()
|
||
|
||
def _load_grants(self):
|
||
if not self._current_user:
|
||
return
|
||
fn = _mysql_get_grants if self._db_type == "mysql" else _pg_get_grants
|
||
self._worker = _UserWorker(
|
||
fn, self._driver, self._current_user, self._current_host, parent=self
|
||
)
|
||
self._worker.result.connect(self._populate_grants)
|
||
self._worker.error.connect(
|
||
lambda e: self._set_status(f"Error loading grants: {e}"))
|
||
self._worker.start()
|
||
|
||
def _populate_grants(self, grants: list[str]):
|
||
t = self._grants_table
|
||
t.setRowCount(len(grants))
|
||
mono = QFont("Consolas", 10)
|
||
for r, g in enumerate(grants):
|
||
item = QTableWidgetItem(g)
|
||
item.setFont(mono)
|
||
t.setItem(r, 0, item)
|
||
self._set_status(
|
||
f"{len(grants)} grant statement(s) for {self._current_user}"
|
||
)
|
||
|
||
# ── CRUD operations ───────────────────────────────────────────────────────
|
||
|
||
def _create_user(self):
|
||
dlg = _CreateUserDialog(self._db_type, parent=self)
|
||
if not dlg.exec():
|
||
return
|
||
fn = _mysql_create_user if self._db_type == "mysql" else _pg_create_user
|
||
self._worker = _UserWorker(
|
||
fn, self._driver, dlg.username, dlg.host, dlg.password, parent=self
|
||
)
|
||
self._worker.result.connect(lambda _: (
|
||
self._set_status(f"User '{dlg.username}' created."),
|
||
self._load_users(),
|
||
))
|
||
self._worker.error.connect(
|
||
lambda e: QMessageBox.critical(self, "Create User Error", str(e))
|
||
)
|
||
self._worker.start()
|
||
|
||
def _drop_user(self):
|
||
if not self._current_user:
|
||
return
|
||
btn = QMessageBox.warning(
|
||
self, "Drop User",
|
||
f"Drop user '{self._current_user}' permanently?\n\n"
|
||
"All associated privileges will be revoked.",
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
QMessageBox.StandardButton.No,
|
||
)
|
||
if btn != QMessageBox.StandardButton.Yes:
|
||
return
|
||
fn = _mysql_drop_user if self._db_type == "mysql" else _pg_drop_user
|
||
self._worker = _UserWorker(
|
||
fn, self._driver, self._current_user, self._current_host, parent=self
|
||
)
|
||
self._worker.result.connect(lambda _: (
|
||
self._set_status(f"User '{self._current_user}' dropped."),
|
||
self._load_users(),
|
||
))
|
||
self._worker.error.connect(
|
||
lambda e: QMessageBox.critical(self, "Drop User Error", str(e))
|
||
)
|
||
self._worker.start()
|
||
|
||
def _grant_privs(self):
|
||
dlg = _GrantDialog(
|
||
self._driver, self._db_type, self._databases,
|
||
self._current_user, self._current_host, mode="grant", parent=self,
|
||
)
|
||
if dlg.exec():
|
||
self._set_status(
|
||
f"Privileges granted to {self._current_user}."
|
||
)
|
||
self._load_grants()
|
||
|
||
def _revoke_privs(self):
|
||
dlg = _GrantDialog(
|
||
self._driver, self._db_type, self._databases,
|
||
self._current_user, self._current_host, mode="revoke", parent=self,
|
||
)
|
||
if dlg.exec():
|
||
self._set_status(
|
||
f"Privileges revoked from {self._current_user}."
|
||
)
|
||
self._load_grants()
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────
|
||
|
||
def _set_status(self, msg: str):
|
||
self._status_lbl.setText(msg)
|
||
self.status_message.emit(msg)
|