Coverage for src/mafw/processor/parameters.py: 99%
181 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Parameter descriptors for the Processor framework.
7This module defines the parameter descriptor system that underpins all
8:class:`~mafw.processor.Processor` subclass configuration. It provides:
10- :class:`~mafw.processor.PassiveParameter` — the private storage backing each parameter.
11- :class:`~mafw.processor.ActiveParameter` — the public descriptor interface for declaring
12 processor parameters in a class body.
13- :func:`ensure_parameter_registration` — a decorator guaranteeing parameters
14 are registered before access.
15- :data:`ParameterType` — a generic type variable binding the parameter value.
16- :data:`_ParameterRegistry` — the ordered mapping of parameter names to their
17 descriptors.
19The helper functions :func:`_copy_parent_parameter_definitions`,
20:func:`_ensure_parameter_definitions`, and :func:`_validate_filter_schema` are
21used by the metaclass to finalise parameter bookkeeping at class-creation time.
23.. versionadded:: 2.3
24 Extracted from the monolithic ``processor.py`` module for improved
25 maintainability and focused testing.
26"""
28from __future__ import annotations
30import inspect
31from collections import OrderedDict
32from collections.abc import Callable, Iterator
33from copy import deepcopy
34from functools import wraps
35from typing import (
36 TYPE_CHECKING,
37 Any,
38 Generic,
39 TypeVar,
40 cast,
41 get_args,
42 get_origin,
43 get_type_hints,
44)
46from mafw.db.db_model import MAFwBaseModel
47from mafw.mafw_errors import ProcessorParameterError
48from mafw.models.filter_schema import FilterSchema
49from mafw.models.parameter_schema import ParameterSchema
51if TYPE_CHECKING:
52 import mafw.processor
54# ---------------------------------------------------------------------------
55# Type variables
56# ---------------------------------------------------------------------------
58ParameterType = TypeVar('ParameterType')
59"""Generic variable type for the :class:`~mafw.processor.ActiveParameter` and :class:`~mafw.processor.PassiveParameter`."""
61_F = TypeVar('_F', bound=Callable[..., Any])
62"""Type variable for generic callable with any return value (private to avoid Sphinx collision)."""
64# ---------------------------------------------------------------------------
65# Type aliases
66# ---------------------------------------------------------------------------
68_ParameterRegistry = OrderedDict[str, 'ActiveParameter[Any]']
69"""Ordered mapping of external parameter names to their :class:`~mafw.processor.ActiveParameter` descriptors."""
72# ---------------------------------------------------------------------------
73# Helper functions (used by ProcessorMeta during class creation)
74# ---------------------------------------------------------------------------
77def _copy_parent_parameter_definitions(owner: type[mafw.processor.Processor]) -> _ParameterRegistry:
78 """
79 Build an ordered dictionary of ActiveParameters inherited from base classes.
81 Walks the MRO (skipping the class itself) and collects all parameter
82 descriptors declared on ancestor classes. This ensures that subclasses
83 inherit parameters defined higher in the hierarchy.
85 :param owner: The Processor subclass whose ancestors are inspected.
86 :return: Aggregated parameter definitions from all base classes.
87 :rtype: _ParameterRegistry
88 """
89 definitions: _ParameterRegistry = OrderedDict()
90 for base in owner.__mro__[1:]:
91 base_definitions = getattr(base, '_parameter_definitions', None)
92 if base_definitions:
93 for name, descriptor in base_definitions.items():
94 definitions[name] = descriptor
95 return definitions
98def _ensure_parameter_definitions(owner: type[mafw.processor.Processor]) -> _ParameterRegistry:
99 """
100 Return the class-level parameter registry, creating it if necessary.
102 If the class does not yet have its own ``_parameter_definitions`` dict
103 (i.e. it has not been touched by the metaclass), one is created by
104 copying definitions from the parent classes.
106 :param owner: The Processor subclass to inspect/initialise.
107 :type owner: type[mafw.processor.Processor]
108 :return: The parameter registry for *owner*.
109 :rtype: _ParameterRegistry
110 """
111 definitions: _ParameterRegistry | None = owner.__dict__.get('_parameter_definitions')
112 if definitions is not None:
113 return definitions
115 definitions = _copy_parent_parameter_definitions(owner)
116 setattr(owner, '_parameter_definitions', definitions)
117 return definitions
120def _validate_filter_schema(owner: type[mafw.processor.Processor]) -> None:
121 """
122 Validate the optional :class:`~mafw.models.filter_schema.FilterSchema` declared on a Processor.
124 Checks that:
126 - ``_filter_schema`` is an instance of :class:`FilterSchema` (if present).
127 - ``root_model`` inherits from :class:`~mafw.db.db_model.MAFwBaseModel`.
128 - All entries in ``allowed_models`` are unique MAFwBaseModel subclasses.
130 :param owner: The Processor subclass being validated.
131 :type owner: type[mafw.processor.Processor]
132 :raises ProcessorParameterError: If any of the above constraints are violated.
133 """
134 schema = getattr(owner, '_filter_schema', None)
135 if schema is None:
136 return
137 if not isinstance(schema, FilterSchema):
138 raise ProcessorParameterError('Processor._filter_schema must be a FilterSchema instance')
140 root_model = schema.root_model
141 if not (inspect.isclass(root_model) and issubclass(root_model, MAFwBaseModel)):
142 raise ProcessorParameterError('FilterSchema.root_model must inherit from MAFwBaseModel')
144 # Collect all models to detect duplicates across root and allowed lists.
145 seen_models = {root_model}
146 for model in schema.allowed_models:
147 if not (inspect.isclass(model) and issubclass(model, MAFwBaseModel)):
148 raise ProcessorParameterError('FilterSchema.allowed_models must contain MAFwBaseModel subclasses')
149 if model in seen_models:
150 raise ProcessorParameterError(f'Model {model!r} already declared in FilterSchema')
151 seen_models.add(model)
154# ---------------------------------------------------------------------------
155# Decorator
156# ---------------------------------------------------------------------------
159def ensure_parameter_registration(func: _F) -> _F:
160 """
161 Decorator to ensure that processor parameters are registered before *func* executes.
163 This is applied to methods that access ``self._processor_parameters`` and need
164 the registration step to have completed first.
166 :param func: The method to wrap.
167 :type func: _F
168 :return: The wrapped method.
169 :rtype: _F
170 :raises ProcessorParameterError: If applied to something that is not a Processor instance method.
171 """
173 @wraps(func)
174 def wrapper(*args: mafw.processor.Processor, **kwargs: Any) -> _F:
175 # Lazy import to avoid circular dependency at module load time.
176 from mafw.processor.base import Processor as _Processor
178 # The first positional argument must be *self* — a Processor instance.
179 if len(args) == 0:
180 raise ProcessorParameterError(
181 'Attempt to apply the ensure_parameter_registration to something different to a Processor subclass.'
182 )
183 self = args[0]
184 if not isinstance(self, _Processor):
185 raise ProcessorParameterError(
186 'Attempt to apply the ensure_parameter_registration to something different to a Processor subclass.'
187 )
188 if self._parameter_registered is False:
189 self._register_parameters()
190 return cast(_F, func(*args, **kwargs))
192 return cast(_F, wrapper)
195# ---------------------------------------------------------------------------
196# PassiveParameter — private storage
197# ---------------------------------------------------------------------------
200class PassiveParameter(Generic[ParameterType]):
201 """
202 An helper class to store processor parameter value and metadata.
204 This class is the private interface used by the :class:`ActiveParameter` descriptor to store its value and metadata.
206 When a new :class:`.ActiveParameter` is added to a class, an instance of a PassiveParameter is added to the
207 processor parameter :attr:`register <.processor.Processor._processor_parameters>`.
209 .. seealso::
211 An explanation on how processor parameters work and should be used is given in :ref:`Understanding processor
212 parameters <parameters>`
214 .. versionchanged:: v2.0.0
216 User should only use :class:`ActiveParameter` and never manually instantiate :class:`PassiveParameter`.
217 """
219 def __init__(
220 self, name: str, value: ParameterType | None = None, default: ParameterType | None = None, help_doc: str = ''
221 ):
222 """
223 Constructor parameters:
225 :param name: The name of the parameter. It must be a valid python identifier.
226 :type name: str
227 :param value: The set value of the parameter. If None, then the default value will be used. Defaults to None.
228 :type value: ParameterType, Optional
229 :param default: The default value for the parameter. It is used if the :attr:`value` is not provided. Defaults to None.
230 :type default: ParameterType, Optional
231 :param help_doc: A brief explanation of the parameter.
232 :type help_doc: str, Optional
233 :raises ProcessorParameterError: if both `value` and `default` are not provided or if `name` is not a valid identifier.
234 """
235 if not name.isidentifier():
236 raise ProcessorParameterError(f'{name} is not a valid python identifier.')
238 self.name = name
240 if value is not None:
241 self._value: ParameterType = value
242 self._is_set = True
243 self._is_optional = False
244 elif default is not None:
245 self._value = default
246 self._is_set = False
247 self._is_optional = True
248 else:
249 raise ProcessorParameterError('Processor parameter cannot have both value and default value set to None')
251 self.doc = help_doc
253 def __rich_repr__(self) -> Iterator[Any]:
254 yield 'name', self.name
255 yield 'value', self.value, None
256 yield 'help_doc', self.doc, ''
258 @property
259 def is_set(self) -> bool:
260 """
261 Property to check if the value has been set.
263 It is useful for optional parameter to see if the current value is the default one, or if the user set it.
264 """
265 return self._is_set
267 @property
268 def value(self) -> ParameterType:
269 """
270 Gets the parameter value.
272 :return: The parameter value.
273 :rtype: ParameterType
274 :raises ProcessorParameterError: if both value and default were not defined.
275 """
276 return self._value
278 @value.setter
279 def value(self, value: ParameterType) -> None:
280 """
281 Sets the parameter value.
283 :param value: The value to be set.
284 :type value: ParameterType
285 """
286 self._value = value
287 self._is_set = True
289 @property
290 def is_optional(self) -> bool:
291 """
292 Property to check if the parameter is optional.
294 :return: True if the parameter is optional
295 :rtype: bool
296 """
297 return self._is_optional
299 def __repr__(self) -> str:
300 args = ['name', 'value', 'doc']
301 values = [getattr(self, arg) for arg in args]
302 return '{klass}({attrs})'.format(
303 klass=self.__class__.__name__,
304 attrs=', '.join(f'{k}={v!r}' for k, v in zip(args, values)),
305 )
308# ---------------------------------------------------------------------------
309# ActiveParameter — public descriptor interface
310# ---------------------------------------------------------------------------
313class ActiveParameter(Generic[ParameterType]):
314 r"""
315 The public interface to the processor parameter.
317 The behaviour of a :class:`.processor.Processor` can be customised by using processor parameters. The value of these
318 parameters can be either set via a configuration file or directly when creating the class.
320 If the user wants to benefit from this facility, they have to add in the instance of the Processor subclass an
321 ActiveParameter instance in this way:
323 .. code-block::
325 class MyProcessor(Processor):
327 # this is the input folder
328 input_folder = ActiveParameter('input_folder', Path(r'C:\'), help_doc='This is where to look for input files')
330 def __init__(self, *args, **kwargs):
331 super().__init(*args, **kwargs)
333 # change the input folder to something else
334 self.input_folder = Path(r'D:\data')
336 # get the value of the parameter
337 print(self.input_folder)
339 The ActiveParameter is a `descriptor <https://docs.python.org/3/glossary.html#term-descriptor>`_, it means that
340 when you create one of them, a lot of work is done behind the scene.
342 In simple words, a processor parameter is made by two objects: a public interface where the user can easily
343 access the value of the parameter and a private interface where all other information (default, documentation...)
344 is also stored.
346 The user does not have to take care of all of this. When a new ActiveParameter instance is added to the class as
347 in the code snippet above, the private interface is automatically created and will stay in the class instance
348 until the end of the class lifetime.
350 To access the private interface, the user can use the :meth:`.processor.Processor.get_parameter` method using the
351 parameter
352 name as a key.
354 The user can assign to an ActiveParameter almost any name. There are just a few invalid parameter names that are
355 used for other purposes. The list of reserved names is available :attr:`here <reserved_names>`. Should the user
356 inadvertently use a reserved named, a :exc:`.ProcessorParameterError` is raised.
358 .. seealso::
360 The private counterpart in the :class:`.processor.PassiveParameter`.
362 An explanation on how processor parameters work and should be used is given in :ref:`Understanding processor
363 parameters <parameters>`
365 The list of :attr:`reserved names <reserved_names>`.
366 """
368 reserved_names: list[str] = ['__logic__', '__filter__', '__new_only__', '__inheritance__', '__enable__']
369 """A list of names that cannot be used as processor parameter names.
371 - `__logic__`
372 - `__filter__`
373 - `__new_only__`
374 - `__inheritance__`
375 - `__enable__`
376 """
378 def __init__(
379 self, name: str, value: ParameterType | None = None, default: ParameterType | None = None, help_doc: str = ''
380 ):
381 """
382 Constructor parameters:
384 :param name: The name of the parameter.
385 :type name: str
386 :param value: The initial value of the parameter. Defaults to None.
387 :type value: ParameterType, Optional
388 :param default: The default value of the parameter, to be used when ``value`` is not set., Defaults to None.
389 :type default: ParameterType, Optional
390 :param help_doc: An explanatory text describing the parameter.
391 :type help_doc: str, Optional
392 """
393 self._value = value
394 self._default = default
395 self._help_doc = help_doc
396 self._external_name = self._validate_name(name)
398 def _validate_name(self, proposed_name: str) -> str:
399 """
400 Validate that the proposed parameter name is not in the list of forbidden names.
402 This private method checks if the provided name is allowed for use as a processor parameter.
403 Names that are listed in :attr:`reserved_names` cannot be used as parameter names.
405 :param proposed_name: The name to be validated for use as a processor parameter.
406 :type proposed_name: str
407 :return: The validated name if it passes the forbidden names check.
408 :rtype: str
409 :raises ProcessorParameterError: If the proposed name is in the list of forbidden names.
410 """
411 if proposed_name not in self.reserved_names:
412 return proposed_name
413 raise ProcessorParameterError(f'Attempt to use a forbidden name ({proposed_name})')
415 def __set_name__(self, owner: type[mafw.processor.Processor], name: str) -> None:
416 self.public_name = name
417 self.private_name = f'param_{name}'
418 self._owner = owner
420 # Register this descriptor in the class-level parameter registry.
421 definitions = _ensure_parameter_definitions(owner)
422 existing = definitions.get(self._external_name)
423 if existing is not None and getattr(existing, '_owner', None) is owner:
424 # Duplicate parameter name on the same class — defer the error until metaclass __init__.
425 if not hasattr(owner, '_parameter_definition_error'): 425 ↛ 431line 425 didn't jump to line 431 because the condition on line 425 was always true
426 setattr(
427 owner,
428 '_parameter_definition_error',
429 ProcessorParameterError(f'Duplicated parameter name ({self._external_name}).'),
430 )
431 return
432 definitions[self._external_name] = self
434 def __get__(
435 self, obj: mafw.processor.Processor, obj_type: type[mafw.processor.Processor]
436 ) -> ActiveParameter[ParameterType] | ParameterType:
437 if obj is None:
438 # Class-level access returns the descriptor itself.
439 return self
441 # Retrieve the instance-level passive parameter holding the current value.
442 param = cast(PassiveParameter[ParameterType], obj._processor_parameters[self._external_name])
443 return param.value
445 def __set__(self, obj: mafw.processor.Processor, value: ParameterType) -> None:
446 param = obj._processor_parameters[self._external_name]
447 param.value = value
449 def to_schema(self) -> ParameterSchema:
450 """
451 Returns the static schema describing this parameter.
453 The schema is derived solely from the descriptor metadata and does not instantiate the owning processor.
454 """
455 annotation = self._resolve_parameter_annotation()
456 default_value = self._schema_default_value()
457 help_text = self._help_doc or None
458 is_list = self._is_list_annotation(annotation, default_value)
459 is_dict = self._is_dict_annotation(annotation, default_value)
460 return ParameterSchema(
461 name=self._external_name,
462 annotation=annotation,
463 default=default_value,
464 help=help_text,
465 is_list=is_list,
466 is_dict=is_dict,
467 )
469 def _resolve_parameter_annotation(self) -> type | None:
470 """Resolve the type annotation for this parameter from the owner class hints."""
471 if not hasattr(self, 'public_name') or getattr(self, '_owner', None) is None:
472 return None
474 target = getattr(self, '_owner', None)
476 try:
477 hints = get_type_hints(target)
478 except Exception:
479 hints = {}
481 hint = hints.get(self.public_name)
482 if hint is None:
483 # Fall back to inferring from the default/value.
484 fallback = self._default if self._default is not None else self._value
485 if fallback is not None:
486 return type(fallback)
487 return None
489 origin = get_origin(hint)
490 if origin is ActiveParameter:
491 args = get_args(hint)
492 if args:
493 return cast(type | None, args[0])
494 return None
495 return cast(type | None, hint)
497 def _schema_default_value(self) -> Any:
498 """Return a safe copy of the default/initial value for schema reporting."""
499 candidate = self._default if self._default is not None else self._value
500 if candidate is None:
501 return None
502 try:
503 return deepcopy(candidate)
504 except Exception:
505 return candidate
507 def _is_list_annotation(self, annotation: type | None, default_value: Any) -> bool:
508 """Check whether the annotation or default value indicates a list type."""
509 return self._matches_container(annotation, default_value, list)
511 def _is_dict_annotation(self, annotation: type | None, default_value: Any) -> bool:
512 """Check whether the annotation or default value indicates a dict type."""
513 return self._matches_container(annotation, default_value, dict)
515 @staticmethod
516 def _matches_container(annotation: type | None, default_value: Any, container: type) -> bool:
517 """Generic check for whether annotation/default matches a given container type."""
518 type_hint = annotation
519 if type_hint is None and default_value is not None:
520 type_hint = type(default_value)
522 if type_hint is None:
523 return False
525 origin = get_origin(type_hint)
526 if origin is container:
527 return True
529 if isinstance(type_hint, type) and issubclass(type_hint, container):
530 return True
532 return False