Source code for mafw.steering_gui.views.storage_editor
# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Editor for named storage volumes in steering files.
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Present controller-fed storage settings and emit typed edit requests.
"""
from __future__ import annotations
import json
import re
import tomllib
from collections.abc import Mapping
from typing import Any
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDialog,
QDialogButtonBox,
QFormLayout,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QSizePolicy,
QTableWidget,
QTableWidgetItem,
QToolButton,
QVBoxLayout,
QWidget,
)
from mafw.steering.models import (
VOLUME_STORAGE_TYPE_SCHEMAS,
StorageAccess,
StorageConfiguration,
StorageVolumeConfiguration,
)
[docs]
class StorageVolumeDialog(QDialog):
"""Create or edit one volume using fields validated before acceptance."""
def __init__(
self,
name: str = '',
volume: StorageVolumeConfiguration | None = None,
*,
existing_names: set[str] | None = None,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
self._existing_names = existing_names or set()
self._original_name = name
self._original_parameters: dict[str, Any] = {}
self.setWindowTitle('Edit storage volume' if volume is not None else 'Add storage volume')
self.resize(520, 420)
layout = QVBoxLayout(self)
form = QFormLayout()
self.name_edit = QLineEdit(name, self)
self.name_edit.setPlaceholderText('e.g. output_vol')
self.root_edit = QLineEdit(volume.root if volume and volume.root is not None else '', self)
self.root_edit.setPlaceholderText('e.g. data/output or ${DATA_ROOT}/output')
self.type_combo = QComboBox(self)
self.type_combo.addItems([item.value for item in VOLUME_STORAGE_TYPE_SCHEMAS])
self.access_combo = QComboBox(self)
self.access_combo.addItems([item.value for item in StorageAccess])
if volume is not None:
backend = str(volume.type)
if self.type_combo.findText(backend) < 0:
self.type_combo.addItem(f'{backend} (unsupported)', backend)
self.type_combo.setCurrentIndex(self.type_combo.count() - 1)
else:
self.type_combo.setCurrentText(backend)
access = str(volume.access)
if self.access_combo.findText(access) < 0:
self.access_combo.addItem(f'{access} (unsupported)', access)
self.access_combo.setCurrentIndex(self.access_combo.count() - 1)
else:
self.access_combo.setCurrentText(access)
self._original_parameters = dict(volume.parameters) if isinstance(volume.parameters, Mapping) else {}
form.addRow('Volume name', self.name_edit)
form.addRow('Root path', self.root_edit)
form.addRow('Backend type', self.type_combo)
form.addRow('Access', self.access_combo)
layout.addLayout(form)
layout.addWidget(QLabel('Backend parameters'))
parameters_row = QHBoxLayout()
self.parameters = QTableWidget(0, 2, self)
self.parameters.setHorizontalHeaderLabels(['Name', 'TOML value'])
self.parameters.horizontalHeader().setStretchLastSection(True)
self.parameters.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
parameters_row.addWidget(self.parameters, 1)
parameter_buttons = QVBoxLayout()
self.add_parameter_button = QToolButton(self)
self.add_parameter_button.setText('+')
self.add_parameter_button.setToolTip('Add a backend parameter')
self.remove_parameter_button = QToolButton(self)
self.remove_parameter_button.setText('−')
self.remove_parameter_button.setToolTip('Remove the selected backend parameter')
parameter_buttons.addWidget(self.add_parameter_button)
parameter_buttons.addWidget(self.remove_parameter_button)
parameter_buttons.addStretch(1)
parameters_row.addLayout(parameter_buttons)
layout.addLayout(parameters_row, 1)
self.error_label = QLabel(self)
self.error_label.setStyleSheet('color: red')
self.error_label.setWordWrap(True)
layout.addWidget(self.error_label)
self.button_box = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel,
parent=self,
)
layout.addWidget(self.button_box)
if volume is not None:
self._populate_parameters(volume.parameters)
# Volume names are stable identifiers and may already be referenced by processors.
self.name_edit.setReadOnly(True)
self.name_edit.setToolTip('Volume names are stable identifiers used by logical file references.')
self.name_edit.textChanged.connect(self._validate_form)
self.root_edit.textChanged.connect(self._validate_form)
self.parameters.itemChanged.connect(lambda _item: self._validate_form())
self.add_parameter_button.clicked.connect(self._add_parameter)
self.remove_parameter_button.clicked.connect(self._remove_parameter)
self.button_box.accepted.connect(self._accept_if_valid)
self.button_box.rejected.connect(self.reject)
self._validate_form()
[docs]
def configuration(self) -> dict[str, Any]:
"""Return the dialog values as typed builder arguments."""
backend = str(self.type_combo.currentData() or self.type_combo.currentText().replace(' (unsupported)', ''))
access = str(self.access_combo.currentData() or self.access_combo.currentText().replace(' (unsupported)', ''))
existing = self._original_parameters.get(backend, {})
parameters: dict[str, Any] = dict(existing) if isinstance(existing, Mapping) else {}
for row in range(self.parameters.rowCount()):
key_item = self.parameters.item(row, 0)
value_item = self.parameters.item(row, 1)
if key_item is None or value_item is None or not key_item.text().strip():
continue
parameters[key_item.text()] = tomllib.loads(f'value = {value_item.text()}')['value']
return {
'root': self.root_edit.text(),
'access': access,
'type': backend,
'parameters': {**self._original_parameters, backend: parameters},
}
[docs]
def _populate_parameters(self, parameters: Mapping[str, Any]) -> None:
"""Load parameters for the selected backend without changing TOML types."""
backend = str(self.type_combo.currentData() or self.type_combo.currentText().replace(' (unsupported)', ''))
values = parameters.get(backend, {})
if isinstance(values, Mapping):
for key, value in values.items():
self._insert_parameter(str(key), _toml_literal(value))
def _insert_parameter(self, name: str = '', value: str = '""') -> None:
row = self.parameters.rowCount()
self.parameters.insertRow(row)
self.parameters.setItem(row, 0, QTableWidgetItem(name))
self.parameters.setItem(row, 1, QTableWidgetItem(value))
[docs]
def _add_parameter(self) -> None:
"""Append an empty parameter row and focus its name cell."""
self._insert_parameter()
self.parameters.setCurrentCell(self.parameters.rowCount() - 1, 0)
item = self.parameters.item(self.parameters.rowCount() - 1, 0)
if item is not None:
self.parameters.editItem(item)
[docs]
def _remove_parameter(self) -> None:
"""Remove all selected parameter rows."""
for row in sorted({index.row() for index in self.parameters.selectionModel().selectedRows()}, reverse=True):
self.parameters.removeRow(row)
[docs]
def _validate_form(self) -> None:
"""Keep OK disabled until required fields and parameters are valid."""
name = self.name_edit.text().strip()
root = self.root_edit.text().strip()
message = ''
if not re.fullmatch(r'[A-Za-z0-9_-]+', name):
message = 'Use letters, numbers, underscores, or hyphens for the volume name.'
elif name != self._original_name and name in self._existing_names:
message = f'A volume named {name!r} already exists.'
elif not root:
message = 'Enter a root path expression.'
else:
try:
self.configuration()
except (tomllib.TOMLDecodeError, ValueError, TypeError) as exc:
message = f'Invalid backend parameter: {exc}'
self.error_label.setText(message)
self.button_box.button(QDialogButtonBox.StandardButton.Ok).setEnabled(not message)
[docs]
def _accept_if_valid(self) -> None:
"""Accept only when local volume fields and TOML parameter values parse."""
self._validate_form()
if self.button_box.button(QDialogButtonBox.StandardButton.Ok).isEnabled():
self.accept()
[docs]
class StorageEditor(QWidget):
"""Edit named volumes and show storage validation feedback.
.. versionadded:: 2.4
"""
enabled_changed = Signal(bool)
volume_added = Signal(str, object)
volumes_removed = Signal(list)
volume_changed = Signal(str, object)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
layout = QVBoxLayout(self)
self.enabled = QCheckBox('Enable storage configuration', self)
layout.addWidget(self.enabled)
content = QHBoxLayout()
self.volume_table = QTableWidget(0, 2, self)
self.volume_table.setHorizontalHeaderLabels(['Volume', 'Root path'])
header = self.volume_table.horizontalHeader()
header.setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
self.volume_table.setTextElideMode(Qt.TextElideMode.ElideRight)
self.volume_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self.volume_table.setSelectionMode(QTableWidget.SelectionMode.ExtendedSelection)
content.addWidget(self.volume_table, 1)
buttons = QVBoxLayout()
self.add_button = QToolButton(self)
self.add_button.setText('Add')
self.add_button.setToolTip('Add a storage volume')
self.remove_button = QToolButton(self)
self.remove_button.setText('Remove')
self.remove_button.setToolTip('Remove the selected storage volumes')
self.edit_button = QToolButton(self)
self.edit_button.setText('Edit')
self.edit_button.setToolTip('Edit the selected storage volume')
for button in (self.add_button, self.remove_button, self.edit_button):
button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
buttons.addWidget(button)
equal_button_width = max(
button.sizeHint().width() for button in (self.add_button, self.remove_button, self.edit_button)
)
for button in (self.add_button, self.remove_button, self.edit_button):
button.setFixedWidth(equal_button_width)
buttons.addStretch(1)
content.addLayout(buttons)
layout.addLayout(content, 1)
self.message = QLabel(self)
self.message.setWordWrap(True)
layout.addWidget(self.message)
self.enabled.toggled.connect(self.enabled_changed)
self.add_button.clicked.connect(self._add_volume)
self.remove_button.clicked.connect(self._remove_selected)
self.edit_button.clicked.connect(self._edit_selected)
self.volume_table.itemSelectionChanged.connect(self._update_selection_actions)
self.volume_table.itemDoubleClicked.connect(lambda _item: self._edit_selected())
self._update_selection_actions()
[docs]
def set_configuration(self, config: StorageConfiguration, *, database_enabled: bool, issues: list[str]) -> None:
"""Refresh volume rows, tooltips, availability, and validation feedback."""
self._configuration = config
self.enabled.blockSignals(True)
self.enabled.setChecked(config.enabled)
self.enabled.blockSignals(False)
self.volume_table.setRowCount(0)
for name, volume in config.volumes.items():
self._add_row(name, volume.root or '')
for name in config.invalid_volumes:
self._add_row(name, 'Invalid volume configuration')
self.enabled.setEnabled(database_enabled)
self.add_button.setEnabled(database_enabled)
self.volume_table.setEnabled(database_enabled and config.enabled)
if not database_enabled:
self.message.setText('Storage is unavailable because DBConfiguration is disabled.')
else:
self.message.setText('\n'.join(issues))
self._update_selection_actions()
[docs]
def _add_row(self, name: str, root: str) -> None:
"""Append one volume name and expose its root through the row tooltip."""
row = self.volume_table.rowCount()
self.volume_table.insertRow(row)
name_item = QTableWidgetItem(name)
name_item.setData(Qt.ItemDataRole.UserRole, name)
self.volume_table.setItem(row, 0, name_item)
root_item = QTableWidgetItem(root)
root_item.setToolTip(root)
self.volume_table.setItem(row, 1, root_item)
[docs]
def _selected_names(self) -> list[str]:
"""Return selected volume names in visual row order."""
rows = sorted({index.row() for index in self.volume_table.selectionModel().selectedRows()})
names: list[str] = []
for row in rows:
item = self.volume_table.item(row, 0)
if item is not None:
names.append(str(item.data(Qt.ItemDataRole.UserRole)))
return names
[docs]
def _update_selection_actions(self) -> None:
"""Enable edit and remove buttons according to the current selection."""
names = self._selected_names()
database_enabled = self.enabled.isEnabled()
storage_active = self.enabled.isChecked()
self.remove_button.setEnabled(database_enabled and storage_active and bool(names))
editable = len(names) == 1 and names[0] in getattr(self, '_configuration', StorageConfiguration()).volumes
self.edit_button.setEnabled(database_enabled and storage_active and editable)
[docs]
def _add_volume(self) -> None:
"""Open the shared volume dialog and emit a validated new volume."""
config = getattr(self, '_configuration', StorageConfiguration())
dialog = StorageVolumeDialog(existing_names=set(config.volumes) | set(config.invalid_volumes), parent=self)
if dialog.exec() == QDialog.DialogCode.Accepted:
self.volume_added.emit(dialog.name_edit.text().strip(), dialog.configuration())
[docs]
def _edit_selected(self) -> None:
"""Edit the selected valid volume and emit its replacement fields."""
names = self._selected_names()
if len(names) != 1:
return
name = names[0]
config = getattr(self, '_configuration', StorageConfiguration())
volume = config.volumes.get(name)
if volume is None:
return
dialog = StorageVolumeDialog(name, volume, existing_names=set(config.volumes), parent=self)
if dialog.exec() == QDialog.DialogCode.Accepted:
self.volume_changed.emit(name, dialog.configuration())
[docs]
def _remove_selected(self) -> None:
"""Remove all selected valid or malformed volume entries."""
names = self._selected_names()
if names:
self.volumes_removed.emit(names)
[docs]
def _toml_literal(value: Any) -> str:
"""Render TOML-compatible values for editing without stringifying their type."""
if isinstance(value, str):
return json.dumps(value, ensure_ascii=False)
if isinstance(value, bool):
return 'true' if value else 'false'
if isinstance(value, list):
return '[' + ', '.join(_toml_literal(item) for item in value) + ']'
if isinstance(value, dict):
return '{' + ', '.join(f'{json.dumps(str(key))} = {_toml_literal(item)}' for key, item in value.items()) + '}'
return str(value)