# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
ProcessorList container for orchestrating sequences of processors.
This module defines :class:`~mafw.processor.ProcessorList`, a specialised list that holds
:class:`~mafw.processor.Processor` instances (or nested ProcessorLists) and
executes them sequentially. It manages shared resources — timer, user interface,
and database connection — distributing them to each processor before execution.
Scientists typically construct a :class:`~mafw.processor.ProcessorList` in their analysis script,
populate it with processor instances representing successive analysis steps, and
call :meth:`~mafw.processor.ProcessorList.execute` to run the full pipeline.
.. versionadded:: 2.3
Extracted from the monolithic ``processor.py`` module for improved
maintainability and focused testing.
"""
from __future__ import annotations
import contextlib
import logging
from collections.abc import Iterable, Mapping
from typing import (
TYPE_CHECKING,
Any,
Self,
SupportsIndex,
)
import peewee
from peewee import Database
# noinspection PyUnresolvedReferences
from playhouse.db_url import connect
from mafw.db.db_connection import build_connection_parameters
from mafw.db.db_model import database_proxy, mafw_model_register
from mafw.enumerators import ProcessorExitStatus
from mafw.mafw_errors import AbortProcessorException, MissingDatabase
from mafw.processor.base import Processor
from mafw.processor.utils import validate_database_conf
from mafw.timer import Timer
from mafw.tools.path_resolution import PathResolutionContext
from mafw.ui.abstract_user_interface import UserInterfaceBase
from mafw.ui.console_user_interface import ConsoleInterface
log = logging.getLogger(__name__)
if TYPE_CHECKING:
import mafw.processor
import mafw.storage
[docs]
class ProcessorList(list['mafw.processor.Processor | ProcessorList']):
"""
A list like collection of processors.
ProcessorList is a subclass of list containing only
:class:`~mafw.processor.Processor` subclasses or other ProcessorLists.
An attempt to add an element that is not a :class:`~mafw.processor.Processor`
or a ProcessorList will raise a TypeError.
Along with an iterable of processors, a new processor list can be built using the following parameters.
A root list initializes its attached storage service after acquiring the
database and creating standard tables. Nested lists and processors borrow
that same initialized service.
.. versionchanged:: 2.4
Added optional runtime storage-service attachment.
"""
def __init__(
self,
*args: mafw.processor.Processor | ProcessorList,
name: str | None = None,
description: str | None = None,
timer: Timer | None = None,
timer_params: dict[str, Any] | None = None,
user_interface: UserInterfaceBase | None = None,
database: Database | None = None,
storage_service: mafw.storage.StorageService | None = None,
database_conf: dict[str, Any] | None = None,
database_environment: Mapping[str, str] | None = None,
create_standard_tables: bool = True,
path_resolution_context: PathResolutionContext | None = None,
path_resolution_managed: bool = False,
):
"""
Constructor parameters:
:param name: The name of the processor list. Defaults to ProcessorList.
:type name: str, Optional
:param description: An optional short description. Default to ProcessorList.
:type description: str, Optional
:param timer: The timer object. If None is provided, a new one will be created. Defaults to None.
:type timer: Timer, Optional
:param timer_params: A dictionary of parameter to build the timer object. Defaults to None.
:type timer_params: dict, Optional
:param user_interface: A user interface. Defaults to None
:type user_interface: UserInterfaceBase, Optional
:param database: A database instance. Defaults to None.
:type database: Database, Optional
:param storage_service: Storage service borrowed or initialized by this list.
:type storage_service: mafw.storage.StorageService | None
:param database_conf: Configuration for the database. Default to None.
:type database_conf: dict, Optional
:param database_environment: Optional environment mapping used to resolve steering database settings.
:type database_environment: Mapping[str, str], Optional
:param create_standard_tables: Whether or not to create the standard tables. Defaults to True.
:type create_standard_tables: bool, Optional
:param path_resolution_context: Owning steering-file context for known path parameters.
:type path_resolution_context: PathResolutionContext, Optional
:param path_resolution_managed: Whether processors without a context are framework-managed.
:type path_resolution_managed: bool
"""
# validate_items takes a tuple of processors, that's why we don't unpack args.
super().__init__(self.validate_items(args))
self._name = name or self.__class__.__name__
self.description = description or self._name
self.timer = timer
self.timer_params = timer_params or {}
self._user_interface = user_interface or ConsoleInterface()
self._resource_stack: contextlib.ExitStack
self._processor_exit_status: ProcessorExitStatus = ProcessorExitStatus.Successful
self.nested_list = False
"""
Boolean flag to identify that this list is actually inside another list.
Similarly to the local resource flag for the :class:`~mafw.processor.Processor`, this flag prevents the user interface to be
added to the resource stack.
"""
# database stuff
self._database: peewee.Database | None = database
self.storage_service = storage_service
"""Optional storage service shared with this list's processors.
Only a root list owns initialization. A nested list must inherit its
service from its root, and conflicting attachments raise ``ValueError``.
"""
self._storage_service_inherited = False
self._database_conf: dict[str, Any] | None = validate_database_conf(database_conf)
self._database_environment = dict(database_environment) if database_environment is not None else None
self.create_standard_tables = create_standard_tables
"""The boolean flag to proceed or skip with standard table creation and initialisation"""
self.path_resolution_context = path_resolution_context
self.path_resolution_managed = path_resolution_managed
for item in self:
self._bind_path_context(item)
self._bind_storage_service(item)
def __setitem__( # type: ignore[override]
self,
__index: SupportsIndex,
__object: mafw.processor.Processor | ProcessorList,
) -> None:
item = self.validate_item(__object)
self._bind_path_context(item)
self._bind_storage_service(item)
super().__setitem__(__index, item)
[docs]
def insert(self, __index: SupportsIndex, __object: mafw.processor.Processor | ProcessorList) -> None:
"""Adds a new processor at the specified index."""
item = self.validate_item(__object)
self._bind_path_context(item)
self._bind_storage_service(item)
super().insert(__index, item)
[docs]
def append(self, __object: mafw.processor.Processor | ProcessorList) -> None:
"""Appends a new processor at the end of the list."""
item = self.validate_item(__object)
self._bind_path_context(item)
self._bind_storage_service(item)
super().append(item)
[docs]
def extend(self, __iterable: Iterable[mafw.processor.Processor | ProcessorList]) -> None:
"""Extends the processor list with a list of processors."""
items = list(__iterable) if __iterable is self else __iterable
for item in items:
self.append(item)
[docs]
@staticmethod
def validate_item(item: mafw.processor.Processor | ProcessorList) -> mafw.processor.Processor | ProcessorList:
"""Validates the item being added."""
if isinstance(item, Processor):
item.local_resource_acquisition = False
return item
elif isinstance(item, ProcessorList):
item.timer_params = dict(suppress_message=True)
item.nested_list = True
return item
else:
raise TypeError(f'Expected Processor or ProcessorList, got {type(item).__name__}')
[docs]
def _bind_path_context(self, item: mafw.processor.Processor | ProcessorList) -> None:
"""Propagate the owning steering context through nested groups."""
if isinstance(item, Processor):
item._set_path_resolution_context(self.path_resolution_context, managed=self.path_resolution_managed)
elif isinstance(item, ProcessorList):
# Mock(spec=ProcessorList) passes validation but does not implement iteration.
try:
children = iter(item)
except TypeError:
return
item.path_resolution_context = self.path_resolution_context
item.path_resolution_managed = self.path_resolution_managed
for child in children:
item._bind_path_context(child)
[docs]
def _bind_storage_service(self, item: mafw.processor.Processor | ProcessorList) -> None:
"""Propagate this list's service without replacing a conflicting attachment."""
if self.storage_service is None:
return
if isinstance(item, Processor):
if item.storage_service is not None and item.storage_service is not self.storage_service:
raise ValueError(f'Processor {item.name!r} already has a different StorageService attached.')
item.storage_service = self.storage_service
elif isinstance(item, ProcessorList):
if item.storage_service is not None and item.storage_service is not self.storage_service:
raise ValueError(f'ProcessorList {item.name!r} already has a different StorageService attached.')
item.storage_service = self.storage_service
item._storage_service_inherited = True
# Nested groups can receive processors after they have been attached.
try:
children = iter(item)
except TypeError:
return
for child in children:
item._bind_storage_service(child)
[docs]
@staticmethod
def validate_items(
items: tuple[mafw.processor.Processor | ProcessorList, ...] = (),
) -> tuple[mafw.processor.Processor | ProcessorList, ...]:
"""Validates a tuple of items being added."""
if not items:
return tuple()
return tuple([ProcessorList.validate_item(item) for item in items if item is not None])
@property
def name(self) -> str:
"""
The name of the processor list
:return: The name of the processor list
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str) -> None:
self._name = name
@property
def processor_exit_status(self) -> ProcessorExitStatus:
"""
The processor exit status.
It refers to the whole processor list execution.
"""
return self._processor_exit_status
@processor_exit_status.setter
def processor_exit_status(self, status: ProcessorExitStatus) -> None:
self._processor_exit_status = status
@property
def database(self) -> peewee.Database:
"""
Returns the database instance
:return: A database instance
:raises MissingDatabase: if a database connection is missing.
"""
if self._database is None:
raise MissingDatabase('Database connection not initialized')
return self._database
[docs]
def execute(self) -> ProcessorExitStatus:
"""
Execute the list of processors.
Similarly to the :class:`.processor.Processor`, ProcessorList can be executed. In simple words, the execute
method of each processor in the list is called exactly in the same sequence as they were added.
"""
with contextlib.ExitStack() as self._resource_stack:
self.acquire_resources()
self._user_interface.create_task(self.name, self.description, completed=0, increment=0, total=len(self))
for i, item in enumerate(self):
if isinstance(item, Processor):
log.info('[bold]Executing [red]%s[/red] processor[/bold]' % item.replica_name)
else:
log.info('[bold]Executing [blue]%s[/blue] processor list[/bold]' % item.name)
self.distribute_resources(item)
item.execute()
self._user_interface.update_task(self.name, increment=1)
self._processor_exit_status = item.processor_exit_status
if self._processor_exit_status == ProcessorExitStatus.Aborted:
msg = 'Processor %s caused the processor list to abort' % item.name
log.error(msg)
raise AbortProcessorException(msg)
self._user_interface.update_task(self.name, completed=len(self), total=len(self))
return self._processor_exit_status
[docs]
def acquire_resources(self) -> None:
"""
Acquires external resources.
The resource acquisition strategy mirrors :meth:`.processor.Processor.acquire_resources`:
if a resource (timer, database) is already provided, it is reused; otherwise
a new one is created and registered with the exit stack for automatic cleanup.
"""
# If we do get resources already active (not None) then we use them,
# otherwise, we create them and add them to the resource stack.
if self.timer is None:
self.timer = self._resource_stack.enter_context(Timer(**self.timer_params))
# The user interface is very likely already initialised by the runner.
# But if this is a nested list, then we must not push the user interface in the stack
# otherwise the user interface context (progress for rich) will be stopped at the end
# of the nested list.
if not self.nested_list:
self._resource_stack.enter_context(self._user_interface)
if self._database is None and self._database_conf is None:
# No database, nor configuration — nothing to do.
pass
elif self._database is None and self._database_conf is not None:
# No database instance, but we have a configuration — create one.
if 'DBConfiguration' in self._database_conf:
conf = self._database_conf['DBConfiguration'] # type1
else:
conf = self._database_conf # type2
db_url, connection_parameters = build_connection_parameters(conf, self._database_environment)
self._database = connect(db_url, **connection_parameters) # type: ignore[no-untyped-call] # playhouse.db_url.connect lacks type stubs
try:
self._database.connect()
self._resource_stack.callback(self._database.close)
except peewee.OperationalError as e:
log.critical('Unable to connect to %s', db_url)
raise e
database_proxy.initialize(self._database)
if self.create_standard_tables:
standard_tables = mafw_model_register.get_standard_tables()
self.database.create_tables(standard_tables)
for table in standard_tables:
table.init()
else: # equiv to if self._database is not None:
# We already have a database — likely inside a nested processor list.
# The connection has been already set and initialised. Nothing else to do.
pass
if self.storage_service is not None:
if self.nested_list:
if not self._storage_service_inherited or not self.storage_service.initialized:
raise RuntimeError(
'A nested ProcessorList cannot own an independent StorageService lifecycle; '
'attach the service to the root ProcessorList.'
)
else:
if self._database is None:
raise RuntimeError(
'A configured StorageService requires a database resource. '
'Provide an active database or DBConfiguration to the root ProcessorList.'
)
# Initialize once for this root execution, after DB binding and standard tables.
self.storage_service.initialize()
[docs]
def distribute_resources(self, processor: mafw.processor.Processor | Self) -> None:
"""Distributes the external resources to the items in the list."""
processor.timer = self.timer
processor._user_interface = self._user_interface
processor._database = self._database
if self.storage_service is not None:
self._bind_storage_service(processor)