301 lines
11 KiB
Python
301 lines
11 KiB
Python
"""
|
|
New / Edit connection dialog.
|
|
Supports MySQL, PostgreSQL, SQLite, and SQL Server.
|
|
"""
|
|
from PyQt6.QtWidgets import (
|
|
QDialog, QDialogButtonBox, QFormLayout, QHBoxLayout, QVBoxLayout,
|
|
QLabel, QLineEdit, QComboBox, QSpinBox, QCheckBox, QPushButton,
|
|
QTabWidget, QWidget, QFileDialog, QMessageBox, QFrame, QColorDialog,
|
|
)
|
|
from PyQt6.QtCore import Qt
|
|
from PyQt6.QtGui import QColor
|
|
|
|
from app.models.connection_model import ConnectionProfile
|
|
from app.config.connections import save_profile
|
|
|
|
|
|
class ColorButton(QPushButton):
|
|
"""A button that shows a solid colour and opens a colour picker."""
|
|
|
|
def __init__(self, color: str = "#89b4fa", parent=None):
|
|
super().__init__(parent)
|
|
self._color = color
|
|
self.setFixedSize(32, 24)
|
|
self._refresh()
|
|
self.clicked.connect(self._pick)
|
|
|
|
def _refresh(self):
|
|
self.setStyleSheet(
|
|
f"background-color:{self._color}; border:1px solid #45475a; border-radius:4px;"
|
|
)
|
|
|
|
def _pick(self):
|
|
col = QColorDialog.getColor(QColor(self._color), self, "Pick a colour")
|
|
if col.isValid():
|
|
self._color = col.name()
|
|
self._refresh()
|
|
|
|
@property
|
|
def color(self) -> str:
|
|
return self._color
|
|
|
|
@color.setter
|
|
def color(self, value: str):
|
|
self._color = value
|
|
self._refresh()
|
|
|
|
|
|
class ConnectionDialog(QDialog):
|
|
"""Dialog for creating or editing a ConnectionProfile."""
|
|
|
|
DB_TYPES = [
|
|
("MySQL", "mysql", 3306),
|
|
("PostgreSQL", "postgresql", 5432),
|
|
("SQLite", "sqlite", 0),
|
|
("SQL Server", "mssql", 1433),
|
|
]
|
|
|
|
def __init__(self, profile: ConnectionProfile = None, parent=None):
|
|
super().__init__(parent)
|
|
self._profile = profile
|
|
self._editing = profile is not None
|
|
self.setWindowTitle("Edit Connection" if self._editing else "New Connection")
|
|
self.setMinimumWidth(520)
|
|
self.setModal(True)
|
|
self._build_ui()
|
|
if self._editing:
|
|
self._populate(profile)
|
|
|
|
# ── UI construction ───────────────────────────────────────────────────────
|
|
|
|
def _build_ui(self):
|
|
root = QVBoxLayout(self)
|
|
root.setSpacing(0)
|
|
|
|
# ── Tabs ──────────────────────────────────────────────────────────────
|
|
tabs = QTabWidget()
|
|
tabs.addTab(self._build_general_tab(), "General")
|
|
tabs.addTab(self._build_ssl_tab(), "SSL / Advanced")
|
|
root.addWidget(tabs)
|
|
|
|
# ── Buttons ───────────────────────────────────────────────────────────
|
|
self._test_btn = QPushButton("Test Connection")
|
|
self._test_btn.clicked.connect(self._test_connection)
|
|
|
|
bbox = QDialogButtonBox(
|
|
QDialogButtonBox.StandardButton.Ok |
|
|
QDialogButtonBox.StandardButton.Cancel
|
|
)
|
|
bbox.accepted.connect(self._accept)
|
|
bbox.rejected.connect(self.reject)
|
|
|
|
btn_row = QHBoxLayout()
|
|
btn_row.addWidget(self._test_btn)
|
|
btn_row.addStretch()
|
|
btn_row.addWidget(bbox)
|
|
root.addSpacing(8)
|
|
root.addLayout(btn_row)
|
|
|
|
def _build_general_tab(self) -> QWidget:
|
|
w = QWidget()
|
|
form = QFormLayout(w)
|
|
form.setRowWrapPolicy(QFormLayout.RowWrapPolicy.DontWrapRows)
|
|
form.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
|
|
form.setSpacing(10)
|
|
form.setContentsMargins(16, 16, 16, 8)
|
|
|
|
# Connection name + colour
|
|
name_row = QHBoxLayout()
|
|
self._name = QLineEdit()
|
|
self._name.setPlaceholderText("My Database")
|
|
self._color_btn = ColorButton()
|
|
name_row.addWidget(self._name, 1)
|
|
name_row.addWidget(self._color_btn)
|
|
form.addRow("Name:", name_row)
|
|
|
|
# DB type selector
|
|
self._db_type = QComboBox()
|
|
for label, _, _ in self.DB_TYPES:
|
|
self._db_type.addItem(label)
|
|
self._db_type.currentIndexChanged.connect(self._on_type_changed)
|
|
form.addRow("Type:", self._db_type)
|
|
|
|
# Separator line
|
|
line = QFrame()
|
|
line.setFrameShape(QFrame.Shape.HLine)
|
|
form.addRow(line)
|
|
|
|
# Host / port
|
|
hp = QHBoxLayout()
|
|
self._host = QLineEdit()
|
|
self._host.setPlaceholderText("localhost")
|
|
self._port = QSpinBox()
|
|
self._port.setRange(1, 65535)
|
|
self._port.setValue(3306)
|
|
self._port.setFixedWidth(90)
|
|
hp.addWidget(self._host, 1)
|
|
hp.addWidget(QLabel("Port:"))
|
|
hp.addWidget(self._port)
|
|
form.addRow("Host:", hp)
|
|
|
|
# Database / file path
|
|
db_row = QHBoxLayout()
|
|
self._database = QLineEdit()
|
|
self._database.setPlaceholderText("database name or file path")
|
|
self._browse_btn = QPushButton("Browse…")
|
|
self._browse_btn.setFixedWidth(80)
|
|
self._browse_btn.clicked.connect(self._browse_file)
|
|
self._browse_btn.setVisible(False)
|
|
db_row.addWidget(self._database, 1)
|
|
db_row.addWidget(self._browse_btn)
|
|
form.addRow("Database:", db_row)
|
|
|
|
# Username / password
|
|
self._username = QLineEdit()
|
|
self._username.setPlaceholderText("username")
|
|
form.addRow("Username:", self._username)
|
|
|
|
self._password = QLineEdit()
|
|
self._password.setPlaceholderText("password")
|
|
self._password.setEchoMode(QLineEdit.EchoMode.Password)
|
|
form.addRow("Password:", self._password)
|
|
|
|
# Timeout
|
|
self._timeout = QSpinBox()
|
|
self._timeout.setRange(1, 300)
|
|
self._timeout.setValue(30)
|
|
self._timeout.setSuffix(" sec")
|
|
form.addRow("Timeout:", self._timeout)
|
|
|
|
return w
|
|
|
|
def _build_ssl_tab(self) -> QWidget:
|
|
w = QWidget()
|
|
form = QFormLayout(w)
|
|
form.setSpacing(10)
|
|
form.setContentsMargins(16, 16, 16, 8)
|
|
|
|
self._ssl = QCheckBox("Use SSL / TLS")
|
|
form.addRow(self._ssl)
|
|
|
|
self._ssl_ca = self._file_row(form, "CA Certificate:")
|
|
self._ssl_cert = self._file_row(form, "Client Certificate:")
|
|
self._ssl_key = self._file_row(form, "Client Key:")
|
|
return w
|
|
|
|
def _file_row(self, form: QFormLayout, label: str) -> QLineEdit:
|
|
row = QHBoxLayout()
|
|
le = QLineEdit()
|
|
le.setPlaceholderText("(optional) path to file")
|
|
btn = QPushButton("…")
|
|
btn.setFixedWidth(32)
|
|
btn.clicked.connect(lambda: self._choose_file(le))
|
|
row.addWidget(le, 1)
|
|
row.addWidget(btn)
|
|
form.addRow(label, row)
|
|
return le
|
|
|
|
# ── Slots ─────────────────────────────────────────────────────────────────
|
|
|
|
def _on_type_changed(self, idx: int):
|
|
_, db_type, default_port = self.DB_TYPES[idx]
|
|
is_sqlite = (db_type == "sqlite")
|
|
self._host.setEnabled(not is_sqlite)
|
|
self._port.setEnabled(not is_sqlite)
|
|
self._username.setEnabled(not is_sqlite)
|
|
self._password.setEnabled(not is_sqlite)
|
|
self._browse_btn.setVisible(is_sqlite)
|
|
if default_port:
|
|
self._port.setValue(default_port)
|
|
|
|
def _browse_file(self):
|
|
path, _ = QFileDialog.getOpenFileName(
|
|
self, "Select SQLite File", "",
|
|
"SQLite Databases (*.db *.sqlite *.sqlite3);;All Files (*)"
|
|
)
|
|
if path:
|
|
self._database.setText(path)
|
|
|
|
def _choose_file(self, target: QLineEdit):
|
|
path, _ = QFileDialog.getOpenFileName(self, "Select File", "", "All Files (*)")
|
|
if path:
|
|
target.setText(path)
|
|
|
|
def _test_connection(self):
|
|
p = self._build_profile()
|
|
from app.drivers import get_driver
|
|
try:
|
|
driver = get_driver(p.db_type, self._driver_config(p))
|
|
ok, msg = driver.test_connection()
|
|
except Exception as e:
|
|
ok, msg = False, str(e)
|
|
|
|
icon = "✅" if ok else "❌"
|
|
QMessageBox.information(self, "Test Connection", f"{icon} {msg}")
|
|
|
|
def _accept(self):
|
|
if not self._name.text().strip():
|
|
QMessageBox.warning(self, "Validation", "Connection name is required.")
|
|
return
|
|
profile = self._build_profile()
|
|
save_profile(profile)
|
|
self._profile = profile
|
|
self.accept()
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _build_profile(self) -> ConnectionProfile:
|
|
idx = self._db_type.currentIndex()
|
|
_, db_type, _ = self.DB_TYPES[idx]
|
|
base = self._profile if self._editing else ConnectionProfile(
|
|
name="", db_type=db_type
|
|
)
|
|
base.name = self._name.text().strip()
|
|
base.db_type = db_type
|
|
base.host = self._host.text().strip()
|
|
base.port = self._port.value()
|
|
base.database = self._database.text().strip()
|
|
base.username = self._username.text().strip()
|
|
base.password = self._password.text()
|
|
base.color = self._color_btn.color
|
|
base.ssl = self._ssl.isChecked()
|
|
base.ssl_ca = self._ssl_ca.text().strip()
|
|
base.ssl_cert = self._ssl_cert.text().strip()
|
|
base.ssl_key = self._ssl_key.text().strip()
|
|
base.connection_timeout = self._timeout.value()
|
|
return base
|
|
|
|
@staticmethod
|
|
def _driver_config(p: ConnectionProfile) -> dict:
|
|
return dict(
|
|
host=p.host, port=p.port, database=p.database,
|
|
user=p.username, password=p.password,
|
|
connection_timeout=p.connection_timeout,
|
|
)
|
|
|
|
def _populate(self, p: ConnectionProfile):
|
|
self._name.setText(p.name)
|
|
self._color_btn.color = p.color
|
|
# Set db type combo
|
|
for i, (_, db_type, _) in enumerate(self.DB_TYPES):
|
|
if db_type == p.db_type:
|
|
self._db_type.setCurrentIndex(i)
|
|
break
|
|
self._host.setText(p.host)
|
|
self._port.setValue(p.port)
|
|
self._database.setText(p.database)
|
|
self._username.setText(p.username)
|
|
self._password.setText(p.password)
|
|
self._timeout.setValue(p.connection_timeout)
|
|
self._ssl.setChecked(p.ssl)
|
|
self._ssl_ca.setText(p.ssl_ca)
|
|
self._ssl_cert.setText(p.ssl_cert)
|
|
self._ssl_key.setText(p.ssl_key)
|
|
self._on_type_changed(self._db_type.currentIndex())
|
|
|
|
# ── Result ────────────────────────────────────────────────────────────────
|
|
|
|
@property
|
|
def profile(self) -> ConnectionProfile:
|
|
return self._profile
|