# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Domain models for steering metadata construction.
:Author: Bulgheroni Antonio
:Description: Simple dataclasses that hold editable steering metadata without execution logic.
"""
from __future__ import annotations
from abc import ABC
from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING, Any, cast
from mafw.db.db_filter import ExprParser
if TYPE_CHECKING:
from mafw.db.db_filter import ExprNode
from mafw.enumerators import LogicalOp
[docs]
class ParameterSource(StrEnum):
"""Origin of a processor parameter value in the GUI pipeline."""
CONFIG = 'config'
INHERITED = 'inherited'
DEFAULT = 'default'
[docs]
class ParameterSchemaStatus(StrEnum):
"""Schema reconciliation status for a processor parameter."""
OK = 'OK'
DEPRECATED = 'Deprecated'
NEW = 'New'
UNKNOWN = 'Unknown'
[docs]
class ProcessorSchemaStatus(StrEnum):
"""Schema reconciliation status for a processor."""
OK = 'OK'
UNKNOWN = 'Unknown'
[docs]
class FilterKind(StrEnum):
"""The kind of filter configuration."""
MODEL = 'model'
FIELD = 'field'
CONDITIONAL = 'conditional'
[docs]
@dataclass
class Condition:
"""A single filter condition."""
operator: str | LogicalOp
value: Any
is_implicit: bool = True
is_valid: bool = True
[docs]
@dataclass
class FieldInfo:
"""Metadata about a model field."""
name: str
type: type
help_text: str | None = None
[docs]
@dataclass
class ModelInfo:
"""Metadata about a database model."""
field_info: dict[str, FieldInfo] = field(default_factory=dict)
[docs]
@dataclass
class FilterConfig(ABC):
"""Base configuration for a filter."""
name: str
kind: FilterKind = field(init=False)
enabled: bool = True
model: str = '' # Common field for all filters
[docs]
@dataclass
class ModelFilterConfig(FilterConfig):
"""
Filter applied to a specific database model.
Each field condition is AND'ed by default unless a custom
logical expression is provided via `logic`.
"""
conditions: dict[str, Condition] = field(default_factory=dict)
logic_str_original: str | None = None
logic_buffer: str | None = None
logic_ast: ExprNode | None = None
logic_dirty: bool = False
logic_is_valid: bool = True
model_info: ModelInfo | None = field(default=None, init=False)
@property
def logic(self) -> str | None:
"""Backward compatibility for logic string."""
return self.logic_str_original
@logic.setter
def logic(self, value: str | None) -> None:
self.logic_str_original = value
if value is not None:
try:
self.logic_ast = ExprParser(value).parse()
except Exception:
self.logic_ast = None
else:
self.logic_ast = None
self.logic_dirty = True
def __post_init__(self) -> None:
self.kind = FilterKind.MODEL
self.fill_model_info()
[docs]
def fill_model_info(self) -> None:
"""Populate model_info by inspecting the database model."""
from mafw.db.db_model import mafw_model_register
if not self.model:
return
try:
model_class = mafw_model_register.get_model(self.model)
except KeyError:
return
self.model_info = ModelInfo()
for field_obj in model_class._meta.sorted_fields: # type: ignore[attr-defined]
f_type = _field_type_from_peewee(field_obj)
f_help = getattr(field_obj, 'help_text', None)
self.model_info.field_info[field_obj.name] = FieldInfo(name=field_obj.name, type=f_type, help_text=f_help)
[docs]
@dataclass
class FieldFilterConfig(FilterConfig):
"""
Filter applied to a single field of a specific model.
Conditions are AND'ed by default unless a custom logical
expression is defined via `logic`.
"""
field_name: str = ''
conditions: dict[str, Condition] = field(default_factory=dict)
logic_str_original: str | None = None
logic_buffer: str | None = None
logic_ast: ExprNode | None = None
logic_dirty: bool = False
logic_is_valid: bool = True
@property
def logic(self) -> str | None:
"""Backward compatibility for logic string."""
return self.logic_str_original
@logic.setter
def logic(self, value: str | None) -> None:
self.logic_str_original = value
if value is not None:
try:
self.logic_ast = ExprParser(value).parse()
except Exception:
self.logic_ast = None
else:
self.logic_ast = None
self.logic_dirty = True
def __post_init__(self) -> None:
self.kind = FilterKind.FIELD
[docs]
@dataclass
class ConditionalFilterConfig(FilterConfig):
"""
Conditional filter applied to a model.
Semantics:
IF condition THEN then_clause ELSE else_clause
"""
condition: Condition | None = None
then_clause: Condition | None = None
else_clause: Condition | None = None
condition_field: str = ''
then_field: str = ''
else_field: str | None = None
auto_named: bool = False
model_info: ModelInfo | None = field(default=None, init=False)
def __post_init__(self) -> None:
self.kind = FilterKind.CONDITIONAL
self.fill_model_info()
[docs]
def is_complete(self) -> bool:
"""Return whether every active clause of the conditional filter is valid."""
if not self.condition or not self.condition.is_valid:
return False
if not self.then_clause or not self.then_clause.is_valid:
return False
if self.else_clause and not self.else_clause.is_valid:
return False
return True
[docs]
def fill_model_info(self) -> None:
"""Populate model_info by inspecting the database model."""
from mafw.db.db_model import mafw_model_register
if not self.model:
return
try:
model_class = mafw_model_register.get_model(self.model)
except KeyError:
return
self.model_info = ModelInfo()
for field_obj in model_class._meta.sorted_fields: # type: ignore[attr-defined]
f_type = _field_type_from_peewee(field_obj)
f_help = getattr(field_obj, 'help_text', None)
self.model_info.field_info[field_obj.name] = FieldInfo(name=field_obj.name, type=f_type, help_text=f_help)
[docs]
@dataclass
class ParameterConfig:
"""Configuration and status for a single processor parameter.
:param name: The name of the parameter.
:param value: The current value of the parameter.
:param default: The default value of the parameter.
:param source: The source of the parameter value.
:param status: The schema reconciliation status of the parameter.
:param help: Documentation text for the parameter.
:param type: The type hint for the parameter.
"""
name: str
value: Any = None
default: Any = None
source: ParameterSource = ParameterSource.CONFIG
status: ParameterSchemaStatus = ParameterSchemaStatus.OK
help: str | None = None
active_override: bool = False
type: Any = None
[docs]
@dataclass
class ProcessorRef:
"""Reference to a processor entry used in processors_to_run lists.
:param base_name: The canonical name of the processor without replica suffix.
:param replica: Optional replica identifier appended to the base name with ``#``.
"""
base_name: str
replica: str | None = None
@property
def full_name(self) -> str:
"""Return the replica-aware full identifier for this reference."""
return f'{self.base_name}#{self.replica}' if self.replica else self.base_name
[docs]
@dataclass
class ProcessorConfig:
"""Configuration store for an individual processor or replica section.
:param name: Replica-aware section name (``Processor`` or ``Processor#Replica``).
:param parameters: Parameter configurations and status.
:param description: Processor documentation shown as a table comment.
:param processor_status: Schema reconciliation status for the processor.
:param filters: Mapping from model name to list of filter configuration objects.
:param logic_str_original: Optional ``__logic__`` string stored in the per-filter table.
:param new_only: Optional ``__new_only__`` override.
:param inheritance: Optional ``__inheritance__`` flag for replicas.
:param has_filter_root: Whether the filter table includes a root entry.
"""
name: str
parameters: dict[str, ParameterConfig] = field(default_factory=dict)
description: str | None = None
processor_status: ProcessorSchemaStatus = ProcessorSchemaStatus.OK
filters: dict[str, list[FilterConfig]] = field(default_factory=dict)
logic_str_original: str | None = None
logic_buffer: str | None = None
logic_ast: ExprNode | None = None
logic_dirty: bool = False
logic_is_valid: bool = True
new_only: bool | None = None
inheritance: bool | None = None
has_filter_root: bool = True
schema_filter_models: list[str] = field(default_factory=list)
@property
def filter_logic(self) -> str | None:
"""Backward compatibility for filter_logic string."""
return self.logic_str_original
@filter_logic.setter
def filter_logic(self, value: str | None) -> None:
self.logic_str_original = value
if value is not None:
try:
self.logic_ast = ExprParser(value).parse()
except Exception:
self.logic_ast = None
else:
self.logic_ast = None
self.logic_dirty = True
[docs]
@dataclass
class GroupConfig:
"""Configuration for a logical processor group section.
:param name: Section name used to reference the group in ``processors_to_run``.
:param processors: Ordered list of processor names and/or nested group names.
:param description: Optional documentation string for the group.
:param attributes: Additional arbitrary attributes for the group.
"""
name: str
processors: list[str]
description: str | None = None
attributes: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass
class GlobalSettings:
"""Top-level globals that appear outside any processor or group table.
:param processors_to_run: Ordered list of processor or group names to execute.
:param analysis_name: Optional name for the analysis.
:param analysis_description: Optional description for the analysis.
:param new_only: Optional flag to process only new data.
:param create_standard_tables: Optional flag to create standard output tables.
:param env_files: Dotenv files used to resolve values for this steering configuration.
"""
processors_to_run: list[str] = field(default_factory=list)
analysis_name: str | None = None
analysis_description: str | None = None
new_only: bool | None = None
create_standard_tables: bool | None = None
env_files: list[str] | None = None
[docs]
@dataclass
class DBConfiguration:
"""Database section content mirrored from the steering file.
:param url: Optional database URL connection string.
:param pragmas: SQLite pragma settings to apply on connection.
:param authentication: Optional authentication subtable for server-based backends.
:param parameters: Optional per-backend configuration parameters.
:param attributes: Additional arbitrary attributes for the database configuration.
:param enabled: Whether ``DBConfiguration`` should be serialized.
"""
url: str | None = None
pragmas: dict[str, Any] = field(default_factory=dict)
authentication: dict[str, Any] = field(default_factory=dict)
parameters: dict[str, dict[str, Any]] = field(default_factory=dict)
attributes: dict[str, Any] = field(default_factory=dict)
enabled: bool = True
@property
def migration_directory(self) -> str | None:
"""The configured schema-migration directory, if any.
This is a first-class accessor over the generic ``attributes`` bag for
the ``migration_directory`` key. Reading and writing it keeps the value
round-tripping through the TOML model like any other database-section
entry.
.. versionadded:: 2.4
:return: The migration directory value, or ``None`` when unset.
:rtype: str | None
"""
value = self.attributes.get('migration_directory')
return str(value) if value is not None else None
@migration_directory.setter
def migration_directory(self, value: str | None) -> None:
if value is None:
self.attributes.pop('migration_directory', None)
else:
self.attributes['migration_directory'] = value
[docs]
class StorageAccess(StrEnum):
"""Access policy configured for a storage volume.
.. versionadded:: 2.4
"""
FULL = 'full'
PROTECTED = 'protected'
READ_ONLY = 'read-only'
[docs]
class VolumeStorageType(StrEnum):
"""Backend type configured for a storage volume.
.. versionadded:: 2.4
"""
FILESYSTEM = 'filesystem'
[docs]
@dataclass(frozen=True)
class VolumeStorageTypeSchema:
"""Describe recognized backend-specific parameter value types.
Unknown parameter names remain preserved by the steering model so files can
round-trip across framework versions. A filesystem volume currently has an
empty schema; future volume types can provide the parameters they recognize.
:param parameter_types: Mapping from recognized parameter names to their
expected Python value types.
.. versionadded:: 2.4
"""
parameter_types: Mapping[str, type] = field(default_factory=dict)
[docs]
def validate(self, parameters: Mapping[str, Any]) -> list[str]:
"""Return declared parameter type errors and leave unknown values intact.
:param parameters: Backend-specific parameter values.
:return: Human-readable issues for recognized parameters with wrong types.
.. versionadded:: 2.4
"""
issues: list[str] = []
for name, expected_type in self.parameter_types.items():
if name in parameters and not isinstance(parameters[name], expected_type):
issues.append(f'parameter {name!r} must be {expected_type.__name__}')
return issues
FILESYSTEM_VOLUME_STORAGE_TYPE_SCHEMA = VolumeStorageTypeSchema()
"""Empty parameter schema for the initial filesystem storage backend.
.. versionadded:: 2.4
"""
VOLUME_STORAGE_TYPE_SCHEMAS: dict[VolumeStorageType, VolumeStorageTypeSchema] = {
VolumeStorageType.FILESYSTEM: FILESYSTEM_VOLUME_STORAGE_TYPE_SCHEMA,
}
"""Schemas available for the storage backends implemented by MAFw.
.. versionadded:: 2.4
"""
[docs]
@dataclass
class StorageVolumeConfiguration:
"""Configured values for one named storage volume.
``root`` stores the literal configured path expression. Runtime owners
resolve it when accessing the filesystem; this model never resolves paths.
:param root: Literal filesystem root expression, or ``None`` when missing.
:param access: Access policy; invalid strings are retained for validation.
:param type: Backend type; invalid strings are retained for validation.
:param parameters: Backend-specific parameters, grouped by backend name.
:param attributes: Additional volume-level TOML values preserved verbatim.
:param invalid_fields: Malformed known fields retained until validation.
.. versionadded:: 2.4
"""
root: str | None = None
access: StorageAccess | str = StorageAccess.FULL
type: VolumeStorageType | str = VolumeStorageType.FILESYSTEM
parameters: dict[str, Any] = field(default_factory=dict)
attributes: dict[str, Any] = field(default_factory=dict)
invalid_fields: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass
class StorageConfiguration:
"""Optional named storage volumes from a steering file.
:param volumes: Mapping of volume names to their configuration.
:param enabled: Whether the TOML section was explicitly present or enabled.
:param invalid_volumes: Malformed volume entries retained for validation
and round-trip serialization.
:param invalid_section: A malformed non-table ``StorageConfiguration`` value.
.. versionadded:: 2.4
"""
volumes: dict[str, StorageVolumeConfiguration] = field(default_factory=dict)
enabled: bool = False
invalid_volumes: dict[str, Any] = field(default_factory=dict)
invalid_section: Any = None
[docs]
@dataclass
class UIConfiguration:
"""User-interface section content mirrored from the steering file.
:param interface: The interface type to use (e.g., ``rich``).
:param attributes: Additional UI settings preserved from the steering file.
"""
interface: str = 'rich'
attributes: dict[str, Any] = field(default_factory=dict)
_FIELD_TYPE_STRINGS: dict[str, type] = {
'INT': int,
'INTEGER': int,
'BIGINT': int,
'SMALLINT': int,
'TINYINT': int,
'FLOAT': float,
'REAL': float,
'DOUBLE': float,
'DECIMAL': float,
'NUMERIC': float,
'CHAR': str,
'VARCHAR': str,
'TEXT': str,
'STRING': str,
'DATE': str,
'DATETIME': str,
'TIME': str,
'BOOLEAN': bool,
'BOOL': bool,
'AUTO': int,
}
"""Mapping from Peewee field_type strings to Python primitives."""
[docs]
def _normalize_field_type(field_type: Any) -> type:
"""Normalize a field descriptor to a supported Python primitive."""
if field_type in (int, float, str, bool):
return cast(type, field_type)
if isinstance(field_type, str):
return _FIELD_TYPE_STRINGS.get(field_type.upper(), str)
return str
[docs]
def _field_type_from_peewee(field_obj: Any) -> type:
"""Extract a primitive type from the given Peewee field definition."""
for attr in ('field_type', 'data_type'):
raw_type = getattr(field_obj, attr, None)
if raw_type is not None:
return _normalize_field_type(raw_type)
return str