Coverage for src/mafw/processor/processor_list.py: 100%
124 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"""
5ProcessorList container for orchestrating sequences of processors.
7This module defines :class:`~mafw.processor.ProcessorList`, a specialised list that holds
8:class:`~mafw.processor.Processor` instances (or nested ProcessorLists) and
9executes them sequentially. It manages shared resources — timer, user interface,
10and database connection — distributing them to each processor before execution.
12Scientists typically construct a :class:`~mafw.processor.ProcessorList` in their analysis script,
13populate it with processor instances representing successive analysis steps, and
14call :meth:`~mafw.processor.ProcessorList.execute` to run the full pipeline.
16.. versionadded:: 2.3
17 Extracted from the monolithic ``processor.py`` module for improved
18 maintainability and focused testing.
19"""
21from __future__ import annotations
23import contextlib
24import logging
25from collections.abc import Iterable
26from typing import (
27 Any,
28 Self,
29 SupportsIndex,
30)
32import peewee
33from peewee import Database
35# noinspection PyUnresolvedReferences
36from playhouse.db_url import connect
38from mafw.db.db_connection import build_connection_parameters
39from mafw.db.db_model import database_proxy, mafw_model_register
40from mafw.enumerators import ProcessorExitStatus
41from mafw.mafw_errors import AbortProcessorException, MissingDatabase
42from mafw.processor.base import Processor
43from mafw.processor.utils import validate_database_conf
44from mafw.timer import Timer
45from mafw.ui.abstract_user_interface import UserInterfaceBase
46from mafw.ui.console_user_interface import ConsoleInterface
48log = logging.getLogger(__name__)
51class ProcessorList(list['Processor | ProcessorList']):
52 """
53 A list like collection of processors.
55 ProcessorList is a subclass of list containing only Processor subclasses or other ProcessorList.
57 An attempt to add an element that is not a Processor or a ProcessorList will raise a TypeError.
59 Along with an iterable of processors, a new processor list can be built using the following parameters.
60 """
62 def __init__(
63 self,
64 *args: Processor | ProcessorList,
65 name: str | None = None,
66 description: str | None = None,
67 timer: Timer | None = None,
68 timer_params: dict[str, Any] | None = None,
69 user_interface: UserInterfaceBase | None = None,
70 database: Database | None = None,
71 database_conf: dict[str, Any] | None = None,
72 create_standard_tables: bool = True,
73 ):
74 """
75 Constructor parameters:
77 :param name: The name of the processor list. Defaults to ProcessorList.
78 :type name: str, Optional
79 :param description: An optional short description. Default to ProcessorList.
80 :type description: str, Optional
81 :param timer: The timer object. If None is provided, a new one will be created. Defaults to None.
82 :type timer: Timer, Optional
83 :param timer_params: A dictionary of parameter to build the timer object. Defaults to None.
84 :type timer_params: dict, Optional
85 :param user_interface: A user interface. Defaults to None
86 :type user_interface: UserInterfaceBase, Optional
87 :param database: A database instance. Defaults to None.
88 :type database: Database, Optional
89 :param database_conf: Configuration for the database. Default to None.
90 :type database_conf: dict, Optional
91 :param create_standard_tables: Whether or not to create the standard tables. Defaults to True.
92 :type create_standard_tables: bool, Optional
93 """
95 # validate_items takes a tuple of processors, that's why we don't unpack args.
96 super().__init__(self.validate_items(args))
97 self._name = name or self.__class__.__name__
98 self.description = description or self._name
100 self.timer = timer
101 self.timer_params = timer_params or {}
102 self._user_interface = user_interface or ConsoleInterface()
104 self._resource_stack: contextlib.ExitStack
105 self._processor_exit_status: ProcessorExitStatus = ProcessorExitStatus.Successful
106 self.nested_list = False
107 """
108 Boolean flag to identify that this list is actually inside another list.
110 Similarly to the local resource flag for the :class:`.base.Processor`, this flag prevent the user interface to be
111 added to the resource stack.
112 """
114 # database stuff
115 self._database: peewee.Database | None = database
116 self._database_conf: dict[str, Any] | None = validate_database_conf(database_conf)
117 self.create_standard_tables = create_standard_tables
118 """The boolean flag to proceed or skip with standard table creation and initialisation"""
120 def __setitem__( # type: ignore[override]
121 self,
122 __index: SupportsIndex,
123 __object: Processor | ProcessorList,
124 ) -> None:
125 super().__setitem__(__index, self.validate_item(__object))
127 def insert(self, __index: SupportsIndex, __object: Processor | ProcessorList) -> None:
128 """Adds a new processor at the specified index."""
129 super().insert(__index, self.validate_item(__object))
131 def append(self, __object: Processor | ProcessorList) -> None:
132 """Appends a new processor at the end of the list."""
133 super().append(self.validate_item(__object))
135 def extend(self, __iterable: Iterable[Processor | ProcessorList]) -> None:
136 """Extends the processor list with a list of processors."""
137 if isinstance(__iterable, type(self)):
138 super().extend(__iterable)
139 else:
140 super().extend([self.validate_item(item) for item in __iterable])
142 @staticmethod
143 def validate_item(item: Processor | ProcessorList) -> Processor | ProcessorList:
144 """Validates the item being added."""
145 if isinstance(item, Processor):
146 item.local_resource_acquisition = False
147 return item
148 elif isinstance(item, ProcessorList):
149 item.timer_params = dict(suppress_message=True)
150 item.nested_list = True
151 return item
152 else:
153 raise TypeError(f'Expected Processor or ProcessorList, got {type(item).__name__}')
155 @staticmethod
156 def validate_items(items: tuple[Processor | ProcessorList, ...] = ()) -> tuple[Processor | ProcessorList, ...]:
157 """Validates a tuple of items being added."""
158 if not items:
159 return tuple()
160 return tuple([ProcessorList.validate_item(item) for item in items if item is not None])
162 @property
163 def name(self) -> str:
164 """
165 The name of the processor list
167 :return: The name of the processor list
168 :rtype: str
169 """
170 return self._name
172 @name.setter
173 def name(self, name: str) -> None:
174 self._name = name
176 @property
177 def processor_exit_status(self) -> ProcessorExitStatus:
178 """
179 The processor exit status.
181 It refers to the whole processor list execution.
182 """
183 return self._processor_exit_status
185 @processor_exit_status.setter
186 def processor_exit_status(self, status: ProcessorExitStatus) -> None:
187 self._processor_exit_status = status
189 @property
190 def database(self) -> peewee.Database:
191 """
192 Returns the database instance
194 :return: A database instance
195 :raises MissingDatabase: if a database connection is missing.
196 """
197 if self._database is None:
198 raise MissingDatabase('Database connection not initialized')
199 return self._database
201 def execute(self) -> ProcessorExitStatus:
202 """
203 Execute the list of processors.
205 Similarly to the :class:`.processor.Processor`, ProcessorList can be executed. In simple words, the execute
206 method of each processor in the list is called exactly in the same sequence as they were added.
207 """
208 with contextlib.ExitStack() as self._resource_stack:
209 self.acquire_resources()
210 self._user_interface.create_task(self.name, self.description, completed=0, increment=0, total=len(self))
211 for i, item in enumerate(self):
212 if isinstance(item, Processor):
213 log.info('[bold]Executing [red]%s[/red] processor[/bold]' % item.replica_name)
214 else:
215 log.info('[bold]Executing [blue]%s[/blue] processor list[/bold]' % item.name)
216 self.distribute_resources(item)
217 item.execute()
218 self._user_interface.update_task(self.name, increment=1)
219 self._processor_exit_status = item.processor_exit_status
220 if self._processor_exit_status == ProcessorExitStatus.Aborted:
221 msg = 'Processor %s caused the processor list to abort' % item.name
222 log.error(msg)
223 raise AbortProcessorException(msg)
224 self._user_interface.update_task(self.name, completed=len(self), total=len(self))
225 return self._processor_exit_status
227 def acquire_resources(self) -> None:
228 """
229 Acquires external resources.
231 The resource acquisition strategy mirrors :meth:`.processor.Processor.acquire_resources`:
232 if a resource (timer, database) is already provided, it is reused; otherwise
233 a new one is created and registered with the exit stack for automatic cleanup.
234 """
235 # If we do get resources already active (not None) then we use them,
236 # otherwise, we create them and add them to the resource stack.
237 if self.timer is None:
238 self.timer = self._resource_stack.enter_context(Timer(**self.timer_params))
239 # The user interface is very likely already initialised by the runner.
240 # But if this is a nested list, then we must not push the user interface in the stack
241 # otherwise the user interface context (progress for rich) will be stopped at the end
242 # of the nested list.
243 if not self.nested_list:
244 self._resource_stack.enter_context(self._user_interface)
245 if self._database is None and self._database_conf is None:
246 # No database, nor configuration — nothing to do.
247 pass
248 elif self._database is None and self._database_conf is not None:
249 # No database instance, but we have a configuration — create one.
250 if 'DBConfiguration' in self._database_conf:
251 conf = self._database_conf['DBConfiguration'] # type1
252 else:
253 conf = self._database_conf # type2
255 db_url, connection_parameters = build_connection_parameters(conf)
257 self._database = connect(db_url, **connection_parameters) # type: ignore[no-untyped-call] # playhouse.db_url.connect lacks type stubs
258 try:
259 self._database.connect()
260 self._resource_stack.callback(self._database.close)
261 except peewee.OperationalError as e:
262 log.critical('Unable to connect to %s', db_url)
263 raise e
264 database_proxy.initialize(self._database)
265 if self.create_standard_tables:
266 standard_tables = mafw_model_register.get_standard_tables()
267 self.database.create_tables(standard_tables)
268 for table in standard_tables:
269 table.init()
270 else: # equiv to if self._database is not None:
271 # We already have a database — likely inside a nested processor list.
272 # The connection has been already set and initialised. Nothing else to do.
273 pass
275 def distribute_resources(self, processor: Processor | Self) -> None:
276 """Distributes the external resources to the items in the list."""
277 processor.timer = self.timer
278 processor._user_interface = self._user_interface
279 processor._database = self._database