# 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 dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING, Any, Dict, List, Optional, 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: Optional[str] = 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: Optional[str] = None
logic_buffer: Optional[str] = None
logic_ast: Optional[ExprNode] = None
logic_dirty: bool = False
logic_is_valid: bool = True
model_info: Optional[ModelInfo] = field(default=None, init=False)
@property
def logic(self) -> Optional[str]:
"""Backward compatibility for logic string."""
return self.logic_str_original
@logic.setter
def logic(self, value: Optional[str]) -> 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: Optional[str] = None
logic_buffer: Optional[str] = None
logic_ast: Optional[ExprNode] = None
logic_dirty: bool = False
logic_is_valid: bool = True
@property
def logic(self) -> Optional[str]:
"""Backward compatibility for logic string."""
return self.logic_str_original
@logic.setter
def logic(self, value: Optional[str]) -> 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: Optional[Condition] = None
then_clause: Optional[Condition] = None
else_clause: Optional[Condition] = None
condition_field: str = ''
then_field: str = ''
else_field: Optional[str] = None
auto_named: bool = False
model_info: Optional[ModelInfo] = 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: Optional[str] = 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 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)
processor_status: ProcessorSchemaStatus = ProcessorSchemaStatus.OK
filters: Dict[str, List[FilterConfig]] = field(default_factory=dict)
logic_str_original: Optional[str] = None
logic_buffer: Optional[str] = None
logic_ast: Optional[ExprNode] = None
logic_dirty: bool = False
logic_is_valid: bool = True
new_only: Optional[bool] = None
inheritance: Optional[bool] = None
has_filter_root: bool = True
schema_filter_models: List[str] = field(default_factory=list)
@property
def filter_logic(self) -> Optional[str]:
"""Backward compatibility for filter_logic string."""
return self.logic_str_original
@filter_logic.setter
def filter_logic(self, value: Optional[str]) -> 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: Optional[str] = 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.
"""
processors_to_run: List[str] = field(default_factory=list)
analysis_name: Optional[str] = None
analysis_description: Optional[str] = None
new_only: Optional[bool] = None
create_standard_tables: Optional[bool] = 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 attributes: Additional arbitrary attributes for the database configuration.
:param enabled: Whether ``DBConfiguration`` should be serialized.
"""
url: Optional[str] = None
pragmas: Dict[str, Any] = field(default_factory=dict)
attributes: Dict[str, Any] = field(default_factory=dict)
enabled: bool = True
[docs]
@dataclass
class UIConfiguration:
"""User-interface section content mirrored from the steering file.
:param interface: The interface type to use (e.g., ``rich``).
"""
interface: str = 'rich'
_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