# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Plugin management system for MAFw framework.
This module provides the core functionality for loading and managing plugins within the MAFw framework.
It supports loading various types of plugins including processors, standard tables, database models, and
user interfaces.
The plugin manager uses the pluggy library to handle plugin discovery and registration
through entry points and hooks.
When plugins are loaded using the :meth:`.MAFwPluginManager.load_plugins` function, the job is divided into multiple
threads to improve performance.
Key features:
- Asynchronous plugin loading with progress indication
- Support for both internal and external plugins
- Type-safe plugin handling with proper data structures
- Logging integration for monitoring plugin loading processes
- Global plugin manager singleton for consistent access
The module defines several key components:
- :class:`.LoadedPlugins`: Data container for loaded plugins
- :class:`.MAFwPluginManager`: Main plugin manager class
- :func:`.get_plugin_manager`: Factory function for accessing the plugin manager
Plugin types supported:
- Processors (`processors`): Classes that implement data processing logic
- Database Modules (`db_modules`): Model modules for database interaction
- User Interfaces (`ui`): UI implementations for different interfaces
.. versionchanged:: v2.0.0
Complete refactoring of the plugin manager system.
"""
import importlib
import itertools
import logging
import time
from collections.abc import Iterable
from concurrent.futures import Future, ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal, cast
import pluggy
from mafw import hookspecs, plugins
from mafw.lazy_import import ProcessorClassProtocol, UserInterfaceClassProtocol
from mafw.mafw_errors import AmbiguousProcessor, UnknownProcessor
if TYPE_CHECKING:
pass
log = logging.getLogger(__name__)
[docs]
def _derive_identity(processor: ProcessorClassProtocol) -> tuple[str, str]:
"""
Return the ``(qualifier, short_name)`` pair identifying a loaded processor.
The *qualifier* is the processor's top-level import package name and the *short name* is its
unqualified class name. Lazy-imported processors expose ``plugin_qualname`` (a dotted
``module.ClassName`` identifier) and ``plugin_name`` (the class name); eagerly registered
``Processor`` subclasses expose only the class dunders. Both cases are handled here so the rule
is defined exactly once.
:param processor: The loaded processor (lazy proxy or eager class).
:type processor: ProcessorClassProtocol
:return: The ``(qualifier, short_name)`` pair.
:rtype: tuple[str, str]
"""
if hasattr(processor, 'plugin_qualname'):
# Lazy import proxy: the qualifier is the top-level package of the dotted qualname.
qualifier = processor.plugin_qualname.split('.', 1)[0]
short_name = processor.plugin_name
else:
# Eagerly registered class: derive the qualifier from the module's top-level package.
qualifier = processor.__module__.split('.', 1)[0]
short_name = processor.__name__
return qualifier, short_name
[docs]
@dataclass
class ProcessorLookup:
"""
Resolve processor references while keeping every discovered processor addressable.
MAFw identifies processors by their ``Short_Name`` (the unqualified class name). When two
installed plugin packages export processors sharing a ``Short_Name`` the legacy
:attr:`LoadedPlugins.processor_dict` kept only one of them, chosen by registration order. This
lookup instead retains **all** candidates: it indexes processors both by ``Short_Name`` (one to
many) and by ``Qualified_Name`` (``package.ShortName``, one to one), so collisions can be
detected and resolved explicitly.
The resolution policy implemented by :meth:`resolve` is:
- a dotted name is a ``Qualified_Name`` and resolves to its single processor or raises
:class:`~mafw.mafw_errors.UnknownProcessor`;
- an undotted name is a ``Short_Name``: a single candidate resolves, two or more raise
:class:`~mafw.mafw_errors.AmbiguousProcessor`, and none raises
:class:`~mafw.mafw_errors.UnknownProcessor`.
.. versionadded:: v2.4
"""
#: short name -> every processor that declares it (collision candidates)
by_short_name: dict[str, list[ProcessorClassProtocol]] = field(default_factory=dict)
#: fully qualified name (``package.ShortName``) -> the single processor providing it
by_qualified_name: dict[str, ProcessorClassProtocol] = field(default_factory=dict)
[docs]
def qualified_name(self, processor: ProcessorClassProtocol) -> str:
"""
Return the ``package.ShortName`` qualified name of a processor.
.. versionadded:: v2.4
:param processor: The processor whose qualified name is requested.
:type processor: ProcessorClassProtocol
:return: The qualified name ``package.ShortName``.
:rtype: str
"""
qualifier, short_name = _derive_identity(processor)
return f'{qualifier}.{short_name}'
[docs]
def is_colliding(self, short_name: str) -> bool:
"""
Return ``True`` when two or more loaded processors share ``short_name``.
.. versionadded:: v2.4
:param short_name: The short name to test.
:type short_name: str
:return: ``True`` if the short name is ambiguous.
:rtype: bool
"""
return len(self.by_short_name.get(short_name, [])) >= 2
[docs]
def alternatives(self, short_name: str) -> list[str]:
"""
Return the sorted qualified names available for a short name.
.. versionadded:: v2.4
:param short_name: The short name whose alternatives are requested.
:type short_name: str
:return: The sorted list of qualified names providing that short name.
:rtype: list[str]
"""
return sorted(self.qualified_name(candidate) for candidate in self.by_short_name.get(short_name, []))
[docs]
def resolve(self, name: str) -> ProcessorClassProtocol:
"""
Resolve a (possibly dotted) processor name to a single processor.
.. versionadded:: v2.4
:param name: A ``Short_Name`` or a ``Qualified_Name`` (``package.ShortName``). The replica
suffix (``#replica``) must already have been stripped by the caller.
:type name: str
:return: The single processor the name resolves to.
:rtype: ProcessorClassProtocol
:raises AmbiguousProcessor: when an undotted short name is shared by two or more processors.
:raises UnknownProcessor: when the name matches no loaded processor.
"""
if '.' in name:
# Qualified name: it must match exactly one registered qualified name.
try:
return self.by_qualified_name[name]
except KeyError:
raise UnknownProcessor(f"Qualified processor name '{name}' matches no loaded processor") from None
candidates = self.by_short_name.get(name, [])
if len(candidates) == 1:
return candidates[0]
if len(candidates) >= 2:
raise AmbiguousProcessor(name, self.alternatives(name))
raise UnknownProcessor(f"Processor name '{name}' matches no loaded processor")
[docs]
def build_processor_lookup(processors: Iterable[ProcessorClassProtocol]) -> 'ProcessorLookup':
"""
Build a :class:`ProcessorLookup` from an iterable of loaded processors.
This helper centralises the index-building rule so the plugin manager and the steering-file
generator share one definition of short names, qualified names, and collisions.
.. versionadded:: v2.4
:param processors: The loaded processors to index.
:type processors: Iterable[ProcessorClassProtocol]
:return: The populated lookup.
:rtype: ProcessorLookup
"""
lookup = ProcessorLookup()
for processor in processors:
qualifier, short_name = _derive_identity(processor)
lookup.by_short_name.setdefault(short_name, []).append(processor)
lookup.by_qualified_name[f'{qualifier}.{short_name}'] = processor
return lookup
[docs]
def build_processor_dict(lookup: 'ProcessorLookup') -> dict[str, ProcessorClassProtocol]:
"""
Build the backward-compatible ``processor_dict`` view from a :class:`ProcessorLookup`.
The view maps every qualified name to its processor plus every **unique** short name to its
single processor; colliding short names are intentionally omitted (see
:attr:`LoadedPlugins.processor_dict`).
.. versionadded:: v2.4
:param lookup: The lookup to derive the compatibility view from.
:type lookup: ProcessorLookup
:return: The ``processor_dict`` mapping.
:rtype: dict[str, ProcessorClassProtocol]
"""
dct: dict[str, ProcessorClassProtocol] = dict(lookup.by_qualified_name)
for short_name, candidates in lookup.by_short_name.items():
if len(candidates) == 1:
dct[short_name] = candidates[0]
return dct
[docs]
@dataclass
class LoadedPlugins:
"""
Container class for storing loaded plugins of various types.
This dataclass holds collections of different plugin types that have been loaded
by the :class:`MAFwPluginManager`. It provides organized storage for processors,
database modules, and user interfaces.
.. versionadded:: v2.0.0
"""
processor_list: list[ProcessorClassProtocol] = field(default_factory=list)
# List[Type['Processor'] | 'LazyPlugin'] = field(default_factory=list)
"""List of loaded processor classes."""
processor_dict: dict[str, ProcessorClassProtocol] = field(default_factory=dict)
# Dict[str, Type['Processor'] | 'LazyPlugin'] = field(default_factory=dict)
"""Dictionary mapping processor names to their classes.
This is a backward-compatible view: it maps every **unique** short name to its single processor
plus every **qualified name** (``package.ShortName``) to its processor. Short names shared by two
or more processors are intentionally omitted so that an ambiguous unqualified reference is no
longer silently resolved to an arbitrary processor. Use :attr:`processor_lookup` for the full,
ambiguity-aware resolution policy.
"""
processor_lookup: ProcessorLookup = field(default_factory=ProcessorLookup)
"""Resolver retaining every discovered processor, indexed by short and qualified name.
.. versionadded:: v2.4
"""
db_model_modules: list[str] = field(default_factory=list)
"""List of database model module names."""
ui_list: list[UserInterfaceClassProtocol] = field(default_factory=list)
# List[Type['UserInterfaceBase'] | 'LazyPlugin'] = field(default_factory=list)
"""List of loaded user interface classes."""
ui_dict: dict[str, UserInterfaceClassProtocol] = field(default_factory=dict)
# Dict[str, Type['UserInterfaceBase'] | 'LazyPlugin'] = field(default_factory=dict)
"""Dictionary mapping user interface names to their classes."""
def __post_init__(self) -> None:
"""Derive the processor lookup from ``processor_list`` when it was not supplied.
This keeps the container internally consistent for callers that build a ``LoadedPlugins``
directly from a ``processor_list`` (for example in tests or ad-hoc runs): the ambiguity-aware
:attr:`processor_lookup` is populated automatically, and the backward-compatible
:attr:`processor_dict` is filled from it when left empty. An explicitly provided lookup is
never overwritten.
.. versionadded:: v2.4
"""
if not self.processor_lookup.by_short_name and not self.processor_lookup.by_qualified_name:
if self.processor_list:
self.processor_lookup = build_processor_lookup(self.processor_list)
if not self.processor_dict:
self.processor_dict = build_processor_dict(self.processor_lookup)
PluginTypes = Literal['processors', 'db_modules', 'ui']
"""Type alias for accepted types of plugins."""
[docs]
def _as_processor_result(obj: object) -> tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]:
"""
Cast an object to the expected processor result type.
This helper function is used to convert the raw result from plugin loading
operations into the expected tuple format for processors.
.. versionadded:: v2.0.0
:param obj: The object to cast
:type obj: object
:return: A tuple containing a list of processor classes and a dictionary mapping
processor names to their classes
:rtype: tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]
"""
return cast(tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]], obj)
[docs]
def _as_ui_result(obj: object) -> tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]:
"""
Cast an object to the expected UI result type.
This helper function is used to convert the raw result from plugin loading
operations into the expected tuple format for user interfaces.
.. versionadded:: v2.0.0
:param obj: The object to cast
:type obj: object
:return: A tuple containing a list of UI classes and a dictionary mapping
UI names to their classes
:rtype: tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]
"""
return cast(tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]], obj)
[docs]
def _as_db_module_result(obj: object) -> list[str]:
"""
Cast an object to the expected database module result type.
This helper function is used to convert the raw result from plugin loading
operations into the expected list format for database modules.
.. versionadded:: v2.0.0
:param obj: The object to cast
:type obj: object
:return: A list of database module names
:rtype: list[str]
"""
return cast(list[str], obj)
[docs]
class MAFwPluginManager(pluggy.PluginManager):
"""
The MAFw plugin manager.
The MAFwPluginManager class manages the loading and registration of plugins within the MAFw framework.
It supports asynchronous loading of various plugin types, including processors, database modules,
and user interfaces, using a thread pool executor for improved performance.
The class provides methods to load each type of plugin and handles delayed status messages if loading takes
longer than expected.
.. versionadded:: v2.0.0
"""
max_loading_delay = 1 # sec
"""
Loading delay before displaying a message.
If the loading of the external plugins is taking more than this value, a message is displayed to inform the user.
"""
def __init__(self, project_name: str = 'mafw'):
super().__init__(project_name)
self._executor = ThreadPoolExecutor(max_workers=4)
[docs]
def load_db_models_plugins(self) -> list[str]:
"""
Load database model modules from the plugin manager.
This method retrieves all database model modules registered through the plugin manager's
:meth:`~mafw.plugins.register_db_model_modules` hook and imports them.
:returns: List of database model module names
:rtype: list[str]
"""
log.debug('Starting database model plugins...')
db_model_module_list = list(itertools.chain(*self.hook.register_db_model_modules()))
for module in db_model_module_list:
importlib.import_module(module)
log.debug('Finished database model plugins')
return db_model_module_list
[docs]
def load_processor_plugins(self) -> tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]:
"""
Load available processor plugins from the plugin manager.
This method retrieves all processor plugins registered through the plugin manager's
:meth:`~mafw.plugins.register_processors` hook and builds the backward-compatible
``processor_dict`` view from a :class:`ProcessorLookup`.
.. versionchanged:: v2.4
The returned ``processor_dict`` is now the compatibility view derived from a
:class:`ProcessorLookup`: it contains every qualified name plus every unique short name,
but omits short names shared by two or more processors.
:returns: A tuple containing:
- List of available processor classes
- Dictionary mapping processor names to their classes
:rtype: tuple[list[type[processor.Processor]], dict[str, type[processor.Processor]]]
"""
log.debug('Starting processor plugins...')
lst = list(itertools.chain(*self.hook.register_processors()))
lookup = build_processor_lookup(lst)
dct = build_processor_dict(lookup)
log.debug('Finished processor plugins')
return lst, dct
[docs]
def load_user_interface_plugins(
self,
) -> tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]:
"""
Load available user interface plugins from the plugin manager.
This method retrieves all user interface plugins registered through the plugin manager's
:meth:`~mafw.plugins.register_user_interfaces` hook.
:returns: A tuple containing:
- List of available user interface classes
- Dictionary mapping user interface names to their classes
:rtype: tuple[list[type[UserInterfaceBase]], dict[str, type[UserInterfaceBase]]]
"""
log.debug('Start loading user interface plugins...')
lst = list(itertools.chain(*self.hook.register_user_interfaces()))
dct = {ui.name: ui for ui in lst}
log.debug('Finished loading user interface plugins')
return lst, dct
[docs]
def _delayed_status_message(self, futures: list[Future[Any]]) -> None:
"""
Display a warning message if plugin loading takes longer than expected.
This method is called after a delay to check if all plugin loading operations
have completed. If not, it logs a warning message to inform the user that
plugin loading is taking longer than expected.
:param futures: List of futures representing ongoing plugin loading operations
:type futures: list[concurrent.futures.Future]
"""
time.sleep(self.max_loading_delay)
if not all(f.done() for f in futures):
log.warning('Plugin loading is taking longer than expected, please be patient.')
[docs]
def load_plugins(self, plugins_to_load: Iterable[PluginTypes]) -> LoadedPlugins:
"""
Load plugins of specified types in multiple threads.
This method loads plugins of the specified types using a thread pool executor
for improved performance. It handles different plugin types including processors,
standard tables, database modules, and user interfaces.
:param plugins_to_load: Iterable of plugin types to load
:type plugins_to_load: Iterable[:obj:`PluginTypes`]
:return: Container with loaded plugins of all requested types
:rtype: :obj:`LoadedPlugins`
"""
plugins_to_load = list(dict.fromkeys(plugins_to_load))
if not plugins_to_load:
return LoadedPlugins()
lut = {
'processors': self.load_processor_plugins,
'db_modules': self.load_db_models_plugins,
'ui': self.load_user_interface_plugins,
}
# drop invalid plugin types
plugins_to_load = [p for p in plugins_to_load if p in lut]
if not plugins_to_load:
return LoadedPlugins()
log.debug(f'Status message will appear if loading takes > {self.max_loading_delay}s')
# Submit tasks to the executor
futures: list[Future[Any]] = []
for plugin_type in plugins_to_load:
fut = self._executor.submit(lut[plugin_type])
futures.append(fut)
# Start delayed status thread
status_thread = self._executor.submit(self._delayed_status_message, futures)
# Wait for completion
results = []
try:
for fut in futures:
results.append(fut.result()) # will re-raise exceptions
finally:
# When all tasks finished, no need for the warning message
if not status_thread.done():
status_thread.cancel()
# Assemble output
plugins_ = LoadedPlugins()
idx = 0
for plugin_type in plugins_to_load:
result = results[idx]
if plugin_type == 'processors':
plugins_.processor_list, plugins_.processor_dict = _as_processor_result(result)
# Build the ambiguity-aware lookup once from the freshly loaded processor list so
# that it travels on LoadedPlugins alongside the compatibility processor_dict.
plugins_.processor_lookup = build_processor_lookup(plugins_.processor_list)
elif plugin_type == 'ui':
plugins_.ui_list, plugins_.ui_dict = _as_ui_result(result)
else: # if plugin_type == 'db_modules':
plugins_.db_model_modules = _as_db_module_result(result)
idx += 1
return plugins_
global_mafw_plugin_manager: dict[str, 'MAFwPluginManager'] = {}
"""The global mafw plugin manager dictionary."""
[docs]
def get_plugin_manager(force_recreate: bool = False) -> 'MAFwPluginManager':
"""
Create a new or return an existing plugin manager for a given project
:param force_recreate: Flag to force the creation of a new plugin manager. Defaults to False
:type force_recreate: bool, Optional
:return: The plugin manager
:rtype: pluggy.PluginManager
"""
if 'mafw' in global_mafw_plugin_manager and force_recreate:
del global_mafw_plugin_manager['mafw']
if 'mafw' not in global_mafw_plugin_manager:
pm = MAFwPluginManager('mafw')
pm.add_hookspecs(hookspecs)
pm.load_setuptools_entrypoints('mafw')
pm.register(plugins)
global_mafw_plugin_manager['mafw'] = pm
return global_mafw_plugin_manager['mafw']