""" 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 = "#8caaee", 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 #51576d; 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) ok_btn = bbox.button(QDialogButtonBox.StandardButton.Ok) if ok_btn: ok_btn.setDefault(True) 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) # Enter in password submits the dialog self._password.returnPressed.connect(self._accept) # Focus starts on the name field self._name.setFocus() 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