Initial Codes
This commit is contained in:
@@ -0,0 +1,631 @@
|
||||
"""
|
||||
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)
|
||||
Reference in New Issue
Block a user