# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Expose the database editor UI for steering file configuration.
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Provide a widget to edit database backend, credentials, and pragmas without touching the builder.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Mapping, cast
from PySide6.QtCore import Signal
from PySide6.QtGui import QStandardItem, QStandardItemModel
from PySide6.QtWidgets import (
QComboBox,
QFileDialog,
QFormLayout,
QGroupBox,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QPushButton,
QTableView,
QToolButton,
QVBoxLayout,
QWidget,
)
from mafw.db.db_configurations import db_scheme, default_conf
from mafw.steering_gui.utils import block_signals
DEFAULT_BACKEND = 'sqlite'
DEFAULT_URLS: dict[str, str] = {backend: url for backend, url in db_scheme.items()}
DEFAULT_PRAGMAS: dict[str, Any] = dict(cast(Mapping[str, Any], default_conf['sqlite']['pragmas']))
[docs]
class DatabaseEditor(QWidget):
"""Widget that exposes database backend selection, credentials, and pragmas."""
configuration_changed = Signal(dict)
test_connection_requested = Signal(dict)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._current_backend = DEFAULT_BACKEND
self.backend_combo = QComboBox()
self.backend_combo.addItems(list(DEFAULT_URLS.keys()))
self.backend_combo.setCurrentText(DEFAULT_BACKEND)
self.url_edit = QLineEdit()
self.url_browse = QToolButton()
self.url_browse.setText('Browse')
self.url_memory = QToolButton()
self.url_memory.setText('Memory')
self.user_edit = QLineEdit()
self.password_file_edit = QLineEdit()
self.password_file_browse = QToolButton()
self.password_file_browse.setText('Browse')
self._pragmas_model = QStandardItemModel(0, 2, self)
self._pragmas_model.setHorizontalHeaderLabels(['Key', 'Value'])
self.pragmas_table = QTableView()
self.pragmas_table.setModel(self._pragmas_model)
self.pragmas_table.horizontalHeader().setStretchLastSection(True)
self.pragmas_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
self.pragmas_table.setSelectionBehavior(QTableView.SelectionBehavior.SelectRows)
self.pragmas_table.setSelectionMode(QTableView.SelectionMode.ExtendedSelection)
self.pragmas_table.verticalHeader().setVisible(False)
self.add_pragma = QToolButton()
self.add_pragma.setText('+')
self.remove_pragma = QToolButton()
self.remove_pragma.setText('−')
self.test_connection_button = QPushButton('Test connection')
self._setup_layout()
self._populate_pragmas(DEFAULT_PRAGMAS)
self._apply_backend(DEFAULT_BACKEND, enforce_default_url=True, active=True)
self._wire_signals()
[docs]
def set_data(self, config: dict[str, Any] | None) -> None:
"""Populate the UI from a configuration dictionary without emitting signals."""
backend = DEFAULT_BACKEND
url_value = ''
user = None
password_file = None
pragmas = dict(DEFAULT_PRAGMAS)
enabled = True
if config is not None:
url_value = config.get('url') or ''
backend_value = config.get('backend')
backend = self._normalize_backend(backend_value) or self._detect_backend_from_url(url_value)
user = config.get('user')
password_file = config.get('read_default_file')
pragmas = config.get('pragmas') or pragmas
enabled = bool(config.get('enabled', True))
if not url_value and backend in DEFAULT_URLS:
url_value = DEFAULT_URLS[backend]
widgets = (
self.backend_combo,
self.url_edit,
self.user_edit,
self.password_file_edit,
self._pragmas_model,
self._group_box,
)
with block_signals(*widgets):
self.backend_combo.setCurrentText(backend)
self.url_edit.setText(url_value)
self.user_edit.setText(user or '')
self.password_file_edit.setText(password_file or '')
self._populate_pragmas(pragmas)
self._group_box.setChecked(enabled)
self._apply_backend(backend, enforce_default_url=False, active=enabled)
[docs]
def get_data(self) -> dict[str, Any]:
"""Return a normalized view of the current database configuration."""
return {
'backend': self._current_backend,
'url': self.url_edit.text().strip() or None,
'user': self._normalize_text(self.user_edit.text()),
'read_default_file': self._normalize_text(self.password_file_edit.text()),
'pragmas': self._pragmas_to_dict(),
'enabled': self._group_box.isChecked(),
}
[docs]
def is_enabled(self) -> bool:
"""Return whether the database configuration group is active."""
return self._group_box.isChecked()
def _setup_layout(self) -> None:
main_layout = QVBoxLayout(self)
self._group_box = QGroupBox('Database configuration')
self._group_box.setCheckable(True)
self._group_box.setChecked(True)
group_layout = QVBoxLayout()
self._group_box.setLayout(group_layout)
main_layout.addWidget(self._group_box)
backend_layout = QFormLayout()
backend_layout.addRow('Backend:', self.backend_combo)
group_layout.addLayout(backend_layout)
url_row = QHBoxLayout()
url_row.addWidget(self.url_edit)
url_row.addWidget(self.url_browse)
url_row.addWidget(self.url_memory)
url_layout = QFormLayout()
url_layout.addRow('URL:', url_row)
group_layout.addLayout(url_layout)
user_layout = QFormLayout()
user_layout.addRow('User:', self.user_edit)
group_layout.addLayout(user_layout)
password_layout = QHBoxLayout()
password_layout.addWidget(self.password_file_edit)
password_layout.addWidget(self.password_file_browse)
password_form = QFormLayout()
password_form.addRow('Password file:', password_layout)
group_layout.addLayout(password_form)
group_layout.addWidget(QLabel('PRAGMAS'))
group_layout.addWidget(self.pragmas_table)
pragma_buttons = QHBoxLayout()
pragma_buttons.addWidget(self.add_pragma)
pragma_buttons.addWidget(self.remove_pragma)
pragma_buttons.addStretch()
group_layout.addLayout(pragma_buttons)
group_layout.addWidget(self.test_connection_button)
def _wire_signals(self) -> None:
self.backend_combo.currentTextChanged.connect(self._on_backend_changed)
self.url_edit.textChanged.connect(self._on_url_changed)
self.url_browse.clicked.connect(self._on_url_browse)
self.url_memory.clicked.connect(self._on_url_memory)
self.user_edit.textChanged.connect(self._emit_configuration_changed)
self.password_file_edit.textChanged.connect(self._emit_configuration_changed)
self.password_file_browse.clicked.connect(self._on_password_file_browse)
self._pragmas_model.itemChanged.connect(self._on_pragmas_item_changed)
self._pragmas_model.rowsInserted.connect(self._emit_configuration_changed)
self._pragmas_model.rowsRemoved.connect(self._emit_configuration_changed)
self.add_pragma.clicked.connect(self._on_add_pragma)
self.remove_pragma.clicked.connect(self._on_remove_pragma)
self.test_connection_button.clicked.connect(self._on_test_connection_clicked)
self._group_box.toggled.connect(self._on_groupbox_toggled)
def _normalize_backend(self, value: str | None) -> str | None:
if not value:
return None
value = value.lower().strip()
return value if value in DEFAULT_URLS else None
def _on_backend_changed(self, backend: str) -> None:
normalized = self._normalize_backend(backend) or DEFAULT_BACKEND
self._apply_backend(normalized, enforce_default_url=True, active=self._group_box.isChecked())
self._emit_configuration_changed()
def _on_url_changed(self, url: str) -> None:
detected = self._detect_backend_from_url(url)
if detected != self._current_backend:
with block_signals(self.backend_combo):
self.backend_combo.setCurrentText(detected)
self._apply_backend(detected, enforce_default_url=False, active=self._group_box.isChecked())
self._emit_configuration_changed()
def _on_url_browse(self) -> None:
path, _ = QFileDialog.getOpenFileName(
self,
'Select SQLite database file',
filter='SQLite Database (*.db);;All Files (*.*)',
)
if not path:
return
sqlite_path = Path(path).resolve()
self.url_edit.setText(f'sqlite:///{sqlite_path.as_posix()}')
def _on_url_memory(self) -> None:
self.url_edit.setText('sqlite:///:memory:')
def _on_password_file_browse(self) -> None:
path, _ = QFileDialog.getOpenFileName(self, 'Select MySQL password file')
if not path:
return
self.password_file_edit.setText(str(Path(path)))
def _on_pragmas_item_changed(self, item: QStandardItem) -> None:
self._emit_configuration_changed()
def _on_add_pragma(self) -> None:
row = self._pragmas_model.rowCount()
self._pragmas_model.insertRow(
row,
[QStandardItem(''), QStandardItem('')],
)
self._emit_configuration_changed()
def _on_remove_pragma(self) -> None:
selection = self.pragmas_table.selectionModel().selectedRows()
rows = sorted({index.row() for index in selection}, reverse=True)
if not rows and self._pragmas_model.rowCount():
rows = [self._pragmas_model.rowCount() - 1]
for row in rows:
self._pragmas_model.removeRow(row)
if rows:
self._emit_configuration_changed()
def _on_test_connection_clicked(self) -> None:
self.test_connection_requested.emit(self.get_data())
def _apply_backend(self, backend: str, *, enforce_default_url: bool, active: bool) -> None:
backend = self._normalize_backend(backend) or DEFAULT_BACKEND
self._current_backend = backend
url_prefix = DEFAULT_URLS.get(backend, DEFAULT_URLS[DEFAULT_BACKEND])
if enforce_default_url:
current = self.url_edit.text().strip()
normalized = current.lower()
if not normalized.startswith(url_prefix):
with block_signals(self.url_edit):
self.url_edit.setText(url_prefix)
is_sqlite = backend == 'sqlite'
is_mysql = backend == 'mysql'
self.backend_combo.setEnabled(active)
self.url_edit.setEnabled(active)
self.url_browse.setEnabled(active and is_sqlite)
self.url_memory.setEnabled(active and is_sqlite)
self.pragmas_table.setEnabled(active and is_sqlite)
self.add_pragma.setEnabled(active and is_sqlite)
self.remove_pragma.setEnabled(active and is_sqlite)
self.user_edit.setEnabled(active and (is_mysql or backend == 'postgresql'))
self.password_file_edit.setEnabled(active and is_mysql)
self.password_file_browse.setEnabled(active and is_mysql)
self.test_connection_button.setEnabled(active)
def _on_groupbox_toggled(self, enabled: bool) -> None:
self._apply_backend(self._current_backend, enforce_default_url=False, active=enabled)
self._emit_configuration_changed()
def _populate_pragmas(self, pragmas: Mapping[str, Any]) -> None:
self._pragmas_model.removeRows(0, self._pragmas_model.rowCount())
source_pragmas = pragmas if pragmas else DEFAULT_PRAGMAS
for key, value in source_pragmas.items():
# row = self._pragmas_model.rowCount()
key_item = QStandardItem(str(key))
value_item = QStandardItem(str(value))
key_item.setEditable(True)
value_item.setEditable(True)
self._pragmas_model.appendRow([key_item, value_item])
def _pragmas_to_dict(self) -> dict[str, Any]:
result: dict[str, Any] = {}
for row in range(self._pragmas_model.rowCount()):
key_item = self._pragmas_model.item(row, 0)
value_item = self._pragmas_model.item(row, 1)
if key_item is None:
continue
key_text = key_item.text().strip()
if not key_text:
continue
value_text = value_item.text().strip() if value_item else ''
result[key_text] = self._normalize_pragma_value(value_text)
return result
def _normalize_pragma_value(self, raw_value: str) -> Any:
if not raw_value:
return raw_value
normalized = raw_value.strip()
if not normalized:
return ''
if normalized.lower() in {'true', 'false'}:
return normalized.lower() == 'true'
for converter in (int, float):
try:
return converter(normalized)
except ValueError:
continue
return normalized
def _detect_backend_from_url(self, url: str | None) -> str:
if not url:
return self._current_backend or DEFAULT_BACKEND
candidate = url.strip().lower()
for backend, prefix in DEFAULT_URLS.items():
if candidate.startswith(prefix):
return backend
return self._current_backend or DEFAULT_BACKEND
def _normalize_text(self, value: str) -> str | None:
text = value.strip()
return text or None
def _emit_configuration_changed(self, *args: Any) -> None:
self.configuration_changed.emit(self.get_data())