# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Parameter descriptors for the Processor framework.
This module defines the parameter descriptor system that underpins all
:class:`~mafw.processor.Processor` subclass configuration. It provides:
- :class:`~mafw.processor.PassiveParameter` — the private storage backing each parameter.
- :class:`~mafw.processor.ActiveParameter` — the public descriptor interface for declaring
processor parameters in a class body.
- :func:`ensure_parameter_registration` — a decorator guaranteeing parameters
are registered before access.
- :data:`ParameterType` — a generic type variable binding the parameter value.
- :data:`_ParameterRegistry` — the ordered mapping of parameter names to their
descriptors.
The helper functions :func:`_copy_parent_parameter_definitions`,
:func:`_ensure_parameter_definitions`, and :func:`_validate_filter_schema` are
used by the metaclass to finalise parameter bookkeeping at class-creation time.
.. versionadded:: 2.3
Extracted from the monolithic ``processor.py`` module for improved
maintainability and focused testing.
"""
from __future__ import annotations
import inspect
from collections import OrderedDict
from collections.abc import Callable, Iterator
from copy import deepcopy
from functools import wraps
from typing import (
TYPE_CHECKING,
Any,
Generic,
TypeVar,
cast,
get_args,
get_origin,
get_type_hints,
)
from mafw.db.db_model import MAFwBaseModel
from mafw.mafw_errors import ProcessorParameterError
from mafw.models.filter_schema import FilterSchema
from mafw.models.parameter_schema import ParameterSchema
if TYPE_CHECKING:
import mafw.processor
# ---------------------------------------------------------------------------
# Type variables
# ---------------------------------------------------------------------------
ParameterType = TypeVar('ParameterType')
"""Generic variable type for the :class:`~mafw.processor.ActiveParameter` and :class:`~mafw.processor.PassiveParameter`."""
_F = TypeVar('_F', bound=Callable[..., Any])
"""Type variable for generic callable with any return value (private to avoid Sphinx collision)."""
# ---------------------------------------------------------------------------
# Type aliases
# ---------------------------------------------------------------------------
_ParameterRegistry = OrderedDict[str, 'ActiveParameter[Any]']
"""Ordered mapping of external parameter names to their :class:`~mafw.processor.ActiveParameter` descriptors."""
# ---------------------------------------------------------------------------
# Helper functions (used by ProcessorMeta during class creation)
# ---------------------------------------------------------------------------
[docs]
def _copy_parent_parameter_definitions(owner: type[mafw.processor.Processor]) -> _ParameterRegistry:
"""
Build an ordered dictionary of ActiveParameters inherited from base classes.
Walks the MRO (skipping the class itself) and collects all parameter
descriptors declared on ancestor classes. This ensures that subclasses
inherit parameters defined higher in the hierarchy.
:param owner: The Processor subclass whose ancestors are inspected.
:return: Aggregated parameter definitions from all base classes.
:rtype: _ParameterRegistry
"""
definitions: _ParameterRegistry = OrderedDict()
for base in owner.__mro__[1:]:
base_definitions = getattr(base, '_parameter_definitions', None)
if base_definitions:
for name, descriptor in base_definitions.items():
definitions[name] = descriptor
return definitions
[docs]
def _ensure_parameter_definitions(owner: type[mafw.processor.Processor]) -> _ParameterRegistry:
"""
Return the class-level parameter registry, creating it if necessary.
If the class does not yet have its own ``_parameter_definitions`` dict
(i.e. it has not been touched by the metaclass), one is created by
copying definitions from the parent classes.
:param owner: The Processor subclass to inspect/initialise.
:type owner: type[mafw.processor.Processor]
:return: The parameter registry for *owner*.
:rtype: _ParameterRegistry
"""
definitions: _ParameterRegistry | None = owner.__dict__.get('_parameter_definitions')
if definitions is not None:
return definitions
definitions = _copy_parent_parameter_definitions(owner)
setattr(owner, '_parameter_definitions', definitions)
return definitions
[docs]
def _validate_filter_schema(owner: type[mafw.processor.Processor]) -> None:
"""
Validate the optional :class:`~mafw.models.filter_schema.FilterSchema` declared on a Processor.
Checks that:
- ``_filter_schema`` is an instance of :class:`FilterSchema` (if present).
- ``root_model`` inherits from :class:`~mafw.db.db_model.MAFwBaseModel`.
- All entries in ``allowed_models`` are unique MAFwBaseModel subclasses.
:param owner: The Processor subclass being validated.
:type owner: type[mafw.processor.Processor]
:raises ProcessorParameterError: If any of the above constraints are violated.
"""
schema = getattr(owner, '_filter_schema', None)
if schema is None:
return
if not isinstance(schema, FilterSchema):
raise ProcessorParameterError('Processor._filter_schema must be a FilterSchema instance')
root_model = schema.root_model
if not (inspect.isclass(root_model) and issubclass(root_model, MAFwBaseModel)):
raise ProcessorParameterError('FilterSchema.root_model must inherit from MAFwBaseModel')
# Collect all models to detect duplicates across root and allowed lists.
seen_models = {root_model}
for model in schema.allowed_models:
if not (inspect.isclass(model) and issubclass(model, MAFwBaseModel)):
raise ProcessorParameterError('FilterSchema.allowed_models must contain MAFwBaseModel subclasses')
if model in seen_models:
raise ProcessorParameterError(f'Model {model!r} already declared in FilterSchema')
seen_models.add(model)
# ---------------------------------------------------------------------------
# Decorator
# ---------------------------------------------------------------------------
[docs]
def ensure_parameter_registration(func: _F) -> _F:
"""
Decorator to ensure that processor parameters are registered before *func* executes.
This is applied to methods that access ``self._processor_parameters`` and need
the registration step to have completed first.
:param func: The method to wrap.
:type func: _F
:return: The wrapped method.
:rtype: _F
:raises ProcessorParameterError: If applied to something that is not a Processor instance method.
"""
@wraps(func)
def wrapper(*args: mafw.processor.Processor, **kwargs: Any) -> _F:
# Lazy import to avoid circular dependency at module load time.
from mafw.processor.base import Processor as _Processor
# The first positional argument must be *self* — a Processor instance.
if len(args) == 0:
raise ProcessorParameterError(
'Attempt to apply the ensure_parameter_registration to something different to a Processor subclass.'
)
self = args[0]
if not isinstance(self, _Processor):
raise ProcessorParameterError(
'Attempt to apply the ensure_parameter_registration to something different to a Processor subclass.'
)
if self._parameter_registered is False:
self._register_parameters()
return cast(_F, func(*args, **kwargs))
return cast(_F, wrapper)
# ---------------------------------------------------------------------------
# PassiveParameter — private storage
# ---------------------------------------------------------------------------
[docs]
class PassiveParameter(Generic[ParameterType]):
"""
An helper class to store processor parameter value and metadata.
This class is the private interface used by the :class:`ActiveParameter` descriptor to store its value and metadata.
When a new :class:`.ActiveParameter` is added to a class, an instance of a PassiveParameter is added to the
processor parameter :attr:`register <.processor.Processor._processor_parameters>`.
.. seealso::
An explanation on how processor parameters work and should be used is given in :ref:`Understanding processor
parameters <parameters>`
.. versionchanged:: v2.0.0
User should only use :class:`ActiveParameter` and never manually instantiate :class:`PassiveParameter`.
"""
def __init__(
self, name: str, value: ParameterType | None = None, default: ParameterType | None = None, help_doc: str = ''
):
"""
Constructor parameters:
:param name: The name of the parameter. It must be a valid python identifier.
:type name: str
:param value: The set value of the parameter. If None, then the default value will be used. Defaults to None.
:type value: ParameterType, Optional
:param default: The default value for the parameter. It is used if the :attr:`value` is not provided. Defaults to None.
:type default: ParameterType, Optional
:param help_doc: A brief explanation of the parameter.
:type help_doc: str, Optional
:raises ProcessorParameterError: if both `value` and `default` are not provided or if `name` is not a valid identifier.
"""
if not name.isidentifier():
raise ProcessorParameterError(f'{name} is not a valid python identifier.')
self.name = name
if value is not None:
self._value: ParameterType = value
self._is_set = True
self._is_optional = False
elif default is not None:
self._value = default
self._is_set = False
self._is_optional = True
else:
raise ProcessorParameterError('Processor parameter cannot have both value and default value set to None')
self.doc = help_doc
def __rich_repr__(self) -> Iterator[Any]:
yield 'name', self.name
yield 'value', self.value, None
yield 'help_doc', self.doc, ''
@property
def is_set(self) -> bool:
"""
Property to check if the value has been set.
It is useful for optional parameter to see if the current value is the default one, or if the user set it.
"""
return self._is_set
@property
def value(self) -> ParameterType:
"""
Gets the parameter value.
:return: The parameter value.
:rtype: ParameterType
:raises ProcessorParameterError: if both value and default were not defined.
"""
return self._value
@value.setter
def value(self, value: ParameterType) -> None:
"""
Sets the parameter value.
:param value: The value to be set.
:type value: ParameterType
"""
self._value = value
self._is_set = True
@property
def is_optional(self) -> bool:
"""
Property to check if the parameter is optional.
:return: True if the parameter is optional
:rtype: bool
"""
return self._is_optional
def __repr__(self) -> str:
args = ['name', 'value', 'doc']
values = [getattr(self, arg) for arg in args]
return '{klass}({attrs})'.format(
klass=self.__class__.__name__,
attrs=', '.join(f'{k}={v!r}' for k, v in zip(args, values)),
)
# ---------------------------------------------------------------------------
# ActiveParameter — public descriptor interface
# ---------------------------------------------------------------------------
[docs]
class ActiveParameter(Generic[ParameterType]):
r"""
The public interface to the processor parameter.
The behaviour of a :class:`.processor.Processor` can be customised by using processor parameters. The value of these
parameters can be either set via a configuration file or directly when creating the class.
If the user wants to benefit from this facility, they have to add in the instance of the Processor subclass an
ActiveParameter instance in this way:
.. code-block::
class MyProcessor(Processor):
# this is the input folder
input_folder = ActiveParameter('input_folder', Path(r'C:\'), help_doc='This is where to look for input files')
def __init__(self, *args, **kwargs):
super().__init(*args, **kwargs)
# change the input folder to something else
self.input_folder = Path(r'D:\data')
# get the value of the parameter
print(self.input_folder)
The ActiveParameter is a `descriptor <https://docs.python.org/3/glossary.html#term-descriptor>`_, it means that
when you create one of them, a lot of work is done behind the scene.
In simple words, a processor parameter is made by two objects: a public interface where the user can easily
access the value of the parameter and a private interface where all other information (default, documentation...)
is also stored.
The user does not have to take care of all of this. When a new ActiveParameter instance is added to the class as
in the code snippet above, the private interface is automatically created and will stay in the class instance
until the end of the class lifetime.
To access the private interface, the user can use the :meth:`.processor.Processor.get_parameter` method using the
parameter
name as a key.
The user can assign to an ActiveParameter almost any name. There are just a few invalid parameter names that are
used for other purposes. The list of reserved names is available :attr:`here <reserved_names>`. Should the user
inadvertently use a reserved named, a :exc:`.ProcessorParameterError` is raised.
.. seealso::
The private counterpart in the :class:`.processor.PassiveParameter`.
An explanation on how processor parameters work and should be used is given in :ref:`Understanding processor
parameters <parameters>`
The list of :attr:`reserved names <reserved_names>`.
"""
reserved_names: list[str] = ['__logic__', '__filter__', '__new_only__', '__inheritance__', '__enable__']
"""A list of names that cannot be used as processor parameter names.
- `__logic__`
- `__filter__`
- `__new_only__`
- `__inheritance__`
- `__enable__`
"""
def __init__(
self, name: str, value: ParameterType | None = None, default: ParameterType | None = None, help_doc: str = ''
):
"""
Constructor parameters:
:param name: The name of the parameter.
:type name: str
:param value: The initial value of the parameter. Defaults to None.
:type value: ParameterType, Optional
:param default: The default value of the parameter, to be used when ``value`` is not set., Defaults to None.
:type default: ParameterType, Optional
:param help_doc: An explanatory text describing the parameter.
:type help_doc: str, Optional
"""
self._value = value
self._default = default
self._help_doc = help_doc
self._external_name = self._validate_name(name)
[docs]
def _validate_name(self, proposed_name: str) -> str:
"""
Validate that the proposed parameter name is not in the list of forbidden names.
This private method checks if the provided name is allowed for use as a processor parameter.
Names that are listed in :attr:`reserved_names` cannot be used as parameter names.
:param proposed_name: The name to be validated for use as a processor parameter.
:type proposed_name: str
:return: The validated name if it passes the forbidden names check.
:rtype: str
:raises ProcessorParameterError: If the proposed name is in the list of forbidden names.
"""
if proposed_name not in self.reserved_names:
return proposed_name
raise ProcessorParameterError(f'Attempt to use a forbidden name ({proposed_name})')
def __set_name__(self, owner: type[mafw.processor.Processor], name: str) -> None:
self.public_name = name
self.private_name = f'param_{name}'
self._owner = owner
# Register this descriptor in the class-level parameter registry.
definitions = _ensure_parameter_definitions(owner)
existing = definitions.get(self._external_name)
if existing is not None and getattr(existing, '_owner', None) is owner:
# Duplicate parameter name on the same class — defer the error until metaclass __init__.
if not hasattr(owner, '_parameter_definition_error'):
setattr(
owner,
'_parameter_definition_error',
ProcessorParameterError(f'Duplicated parameter name ({self._external_name}).'),
)
return
definitions[self._external_name] = self
def __get__(
self, obj: mafw.processor.Processor, obj_type: type[mafw.processor.Processor]
) -> ActiveParameter[ParameterType] | ParameterType:
if obj is None:
# Class-level access returns the descriptor itself.
return self
# Retrieve the instance-level passive parameter holding the current value.
param = cast(PassiveParameter[ParameterType], obj._processor_parameters[self._external_name])
return param.value
def __set__(self, obj: mafw.processor.Processor, value: ParameterType) -> None:
param = obj._processor_parameters[self._external_name]
param.value = value
[docs]
def to_schema(self) -> ParameterSchema:
"""
Returns the static schema describing this parameter.
The schema is derived solely from the descriptor metadata and does not instantiate the owning processor.
"""
annotation = self._resolve_parameter_annotation()
default_value = self._schema_default_value()
help_text = self._help_doc or None
is_list = self._is_list_annotation(annotation, default_value)
is_dict = self._is_dict_annotation(annotation, default_value)
return ParameterSchema(
name=self._external_name,
annotation=annotation,
default=default_value,
help=help_text,
is_list=is_list,
is_dict=is_dict,
)
[docs]
def _resolve_parameter_annotation(self) -> type | None:
"""Resolve the type annotation for this parameter from the owner class hints."""
if not hasattr(self, 'public_name') or getattr(self, '_owner', None) is None:
return None
target = getattr(self, '_owner', None)
try:
hints = get_type_hints(target)
except Exception:
hints = {}
hint = hints.get(self.public_name)
if hint is None:
# Fall back to inferring from the default/value.
fallback = self._default if self._default is not None else self._value
if fallback is not None:
return type(fallback)
return None
origin = get_origin(hint)
if origin is ActiveParameter:
args = get_args(hint)
if args:
return cast(type | None, args[0])
return None
return cast(type | None, hint)
[docs]
def _schema_default_value(self) -> Any:
"""Return a safe copy of the default/initial value for schema reporting."""
candidate = self._default if self._default is not None else self._value
if candidate is None:
return None
try:
return deepcopy(candidate)
except Exception:
return candidate
[docs]
def _is_list_annotation(self, annotation: type | None, default_value: Any) -> bool:
"""Check whether the annotation or default value indicates a list type."""
return self._matches_container(annotation, default_value, list)
[docs]
def _is_dict_annotation(self, annotation: type | None, default_value: Any) -> bool:
"""Check whether the annotation or default value indicates a dict type."""
return self._matches_container(annotation, default_value, dict)
[docs]
@staticmethod
def _matches_container(annotation: type | None, default_value: Any, container: type) -> bool:
"""Generic check for whether annotation/default matches a given container type."""
type_hint = annotation
if type_hint is None and default_value is not None:
type_hint = type(default_value)
if type_hint is None:
return False
origin = get_origin(type_hint)
if origin is container:
return True
if isinstance(type_hint, type) and issubclass(type_hint, container):
return True
return False