# 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
from typing import (
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.ui.abstract_user_interface import UserInterfaceBase
from mafw.ui.console_user_interface import ConsoleInterface
log = logging.getLogger(__name__)
[docs]
class ProcessorList(list['Processor | ProcessorList']):
"""
A list like collection of processors.
ProcessorList is a subclass of list containing only Processor subclasses or other ProcessorList.
An attempt to add an element that is not a 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.
"""
def __init__(
self,
*args: 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,
database_conf: dict[str, Any] | None = None,
create_standard_tables: bool = True,
):
"""
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 database_conf: Configuration for the database. Default to None.
:type database_conf: dict, Optional
:param create_standard_tables: Whether or not to create the standard tables. Defaults to True.
:type create_standard_tables: bool, Optional
"""
# 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:`.base.Processor`, this flag prevent the user interface to be
added to the resource stack.
"""
# database stuff
self._database: peewee.Database | None = database
self._database_conf: dict[str, Any] | None = validate_database_conf(database_conf)
self.create_standard_tables = create_standard_tables
"""The boolean flag to proceed or skip with standard table creation and initialisation"""
def __setitem__( # type: ignore[override]
self,
__index: SupportsIndex,
__object: Processor | ProcessorList,
) -> None:
super().__setitem__(__index, self.validate_item(__object))
[docs]
def insert(self, __index: SupportsIndex, __object: Processor | ProcessorList) -> None:
"""Adds a new processor at the specified index."""
super().insert(__index, self.validate_item(__object))
[docs]
def append(self, __object: Processor | ProcessorList) -> None:
"""Appends a new processor at the end of the list."""
super().append(self.validate_item(__object))
[docs]
def extend(self, __iterable: Iterable[Processor | ProcessorList]) -> None:
"""Extends the processor list with a list of processors."""
if isinstance(__iterable, type(self)):
super().extend(__iterable)
else:
super().extend([self.validate_item(item) for item in __iterable])
[docs]
@staticmethod
def validate_item(item: Processor | ProcessorList) -> 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]
@staticmethod
def validate_items(items: tuple[Processor | ProcessorList, ...] = ()) -> tuple[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 = 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
[docs]
def distribute_resources(self, 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