mafw.plugin_manager

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 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:

• LoadedPlugins: Data container for loaded plugins

• MAFwPluginManager: Main plugin manager class

• 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

Changed in version v2.0.0: Complete refactoring of the plugin manager system.

Module Attributes

PluginTypes	Type alias for accepted types of plugins.
global_mafw_plugin_manager	The global mafw plugin manager dictionary.

Functions

build_processor_dict(lookup)	Build the backward-compatible processor_dict view from a ProcessorLookup.
build_processor_lookup(processors)	Build a ProcessorLookup from an iterable of loaded processors.
get_plugin_manager([force_recreate])	Create a new or return an existing plugin manager for a given project

Classes

LoadedPlugins(processor_list, ...)	Container class for storing loaded plugins of various types.
MAFwPluginManager([project_name])	The MAFw plugin manager.
ProcessorLookup(by_short_name, ...)	Resolve processor references while keeping every discovered processor addressable.

class mafw.plugin_manager.LoadedPlugins(processor_list: list[~mafw.lazy_import.ProcessorClassProtocol] = <factory>, processor_dict: dict[str, ~mafw.lazy_import.ProcessorClassProtocol] = <factory>, processor_lookup: ~mafw.plugin_manager.ProcessorLookup = <factory>, db_model_modules: list[str] = <factory>, ui_list: list[~mafw.lazy_import.UserInterfaceClassProtocol] = <factory>, ui_dict: dict[str, ~mafw.lazy_import.UserInterfaceClassProtocol] = <factory>)

Bases: object

Container class for storing loaded plugins of various types.

This dataclass holds collections of different plugin types that have been loaded by the MAFwPluginManager. It provides organized storage for processors, database modules, and user interfaces.

Added in version v2.0.0.

db_model_modules: list[str]

List of database model module names.

processor_dict: dict[str, ProcessorClassProtocol]

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 processor_lookup for the full, ambiguity-aware resolution policy.

processor_list: list[ProcessorClassProtocol]

List of loaded processor classes.

processor_lookup: ProcessorLookup

Resolver retaining every discovered processor, indexed by short and qualified name.

Added in version v2.4.

ui_dict: dict[str, UserInterfaceClassProtocol]

Dictionary mapping user interface names to their classes.

ui_list: list[UserInterfaceClassProtocol]

List of loaded user interface classes.

class mafw.plugin_manager.MAFwPluginManager(project_name: str = 'mafw')

Bases: 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.

Added in version v2.0.0.

_delayed_status_message(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.

Parameters:

futures (list[concurrent.futures.Future]) – List of futures representing ongoing plugin loading operations

load_db_models_plugins() → list[str]

Load database model modules from the plugin manager.

This method retrieves all database model modules registered through the plugin manager’s register_db_model_modules() hook and imports them.

Returns:

List of database model module names

Return type:

list[str]

load_plugins(plugins_to_load: Iterable[Literal['processors', 'db_modules', 'ui']]) → 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.

Parameters:

plugins_to_load (Iterable[PluginTypes]) – Iterable of plugin types to load

Returns:

Container with loaded plugins of all requested types

Return type:

LoadedPlugins

load_processor_plugins() → 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 register_processors() hook and builds the backward-compatible processor_dict view from a ProcessorLookup.

Changed in version v2.4: The returned processor_dict is now the compatibility view derived from a 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

Return type:

tuple[list[type[processor.Processor]], dict[str, type[processor.Processor]]]

load_user_interface_plugins() → 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 register_user_interfaces() hook.

Returns:

A tuple containing: - List of available user interface classes - Dictionary mapping user interface names to their classes

Return type:

tuple[list[type[UserInterfaceBase]], dict[str, type[UserInterfaceBase]]]

max_loading_delay = 1

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.

class mafw.plugin_manager.ProcessorLookup(by_short_name: dict[str, list[~mafw.lazy_import.ProcessorClassProtocol]] = <factory>, by_qualified_name: dict[str, ~mafw.lazy_import.ProcessorClassProtocol] = <factory>)

Bases: object

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 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 resolve() is:

• a dotted name is a Qualified_Name and resolves to its single processor or raises UnknownProcessor;

• an undotted name is a Short_Name: a single candidate resolves, two or more raise AmbiguousProcessor, and none raises UnknownProcessor.

Added in version v2.4.

alternatives(short_name: str) → list[str]

Return the sorted qualified names available for a short name.

Added in version v2.4.

Parameters:

short_name (str) – The short name whose alternatives are requested.

Returns:

The sorted list of qualified names providing that short name.

Return type:

list[str]

is_colliding(short_name: str) → bool

Return True when two or more loaded processors share short_name.

Added in version v2.4.

Parameters:

short_name (str) – The short name to test.

Returns:

True if the short name is ambiguous.

Return type:

bool

qualified_name(processor: ProcessorClassProtocol) → str

Return the package.ShortName qualified name of a processor.

Added in version v2.4.

Parameters:

processor (ProcessorClassProtocol) – The processor whose qualified name is requested.

Returns:

The qualified name package.ShortName.

Return type:

str

resolve(name: str) → ProcessorClassProtocol

Resolve a (possibly dotted) processor name to a single processor.

Added in version v2.4.

Parameters:

name (str) – A Short_Name or a Qualified_Name (package.ShortName). The replica suffix (#replica) must already have been stripped by the caller.

Returns:

The single processor the name resolves to.

Return type:

ProcessorClassProtocol

Raises:

• AmbiguousProcessor – when an undotted short name is shared by two or more processors.

• UnknownProcessor – when the name matches no loaded processor.

by_qualified_name: dict[str, ProcessorClassProtocol]

fully qualified name (package.ShortName) -> the single processor providing it

by_short_name: dict[str, list[ProcessorClassProtocol]]

short name -> every processor that declares it (collision candidates)

mafw.plugin_manager._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.

Added in version v2.0.0.

Parameters:

obj (object) – The object to cast

Returns:

A list of database module names

Return type:

list[str]

mafw.plugin_manager._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.

Added in version v2.0.0.

Parameters:

obj (object) – The object to cast

Returns:

A tuple containing a list of processor classes and a dictionary mapping processor names to their classes

Return type:

tuple[list[ProcessorClassProtocol], dict[str, ProcessorClassProtocol]]

mafw.plugin_manager._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.

Added in version v2.0.0.

Parameters:

obj (object) – The object to cast

Returns:

A tuple containing a list of UI classes and a dictionary mapping UI names to their classes

Return type:

tuple[list[UserInterfaceClassProtocol], dict[str, UserInterfaceClassProtocol]]

mafw.plugin_manager._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.

Parameters:

processor (ProcessorClassProtocol) – The loaded processor (lazy proxy or eager class).

Returns:

The (qualifier, short_name) pair.

Return type:

tuple[str, str]

mafw.plugin_manager.build_processor_dict(lookup: ProcessorLookup) → dict[str, ProcessorClassProtocol]

Build the backward-compatible processor_dict view from a 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 LoadedPlugins.processor_dict).

Added in version v2.4.

Parameters:

lookup (ProcessorLookup) – The lookup to derive the compatibility view from.

Returns:

The processor_dict mapping.

Return type:

dict[str, ProcessorClassProtocol]

mafw.plugin_manager.build_processor_lookup(processors: Iterable[ProcessorClassProtocol]) → ProcessorLookup

Build a 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.

Added in version v2.4.

Parameters:

processors (Iterable[ProcessorClassProtocol]) – The loaded processors to index.

Returns:

The populated lookup.

Return type:

ProcessorLookup

mafw.plugin_manager.get_plugin_manager(force_recreate: bool = False) → MAFwPluginManager

Create a new or return an existing plugin manager for a given project

Parameters:

force_recreate (bool, Optional) – Flag to force the creation of a new plugin manager. Defaults to False

Returns:

The plugin manager

Return type:

pluggy.PluginManager

mafw.plugin_manager.PluginTypes

Type alias for accepted types of plugins.

alias of Literal[‘processors’, ‘db_modules’, ‘ui’]

mafw.plugin_manager.global_mafw_plugin_manager: dict[str, MAFwPluginManager] = {}

The global mafw plugin manager dictionary.