mafw.runner

Provides a container to run configurable and modular analytical tasks.

Classes

MAFwApplication([steering_file, ...])

The MAFw Application.

Bases: object

The MAFw Application.

This class takes care of reading a steering file and from the information retrieved construct a ProcessorList from the processor listed there and execute it.

It is very practical because any combination of processors can be run without having to write dedicated scripts but simply modifying a steering file.

The application will search for processors not only among the ones available in the MAFw library, but also in all other packages exposing processors via the plugin mechanism.

All parameters in the constructor are optional.

An instance can be created also without the steering_file, but such an instance cannot be executed. The steering file can be provided in a later stage via the init() method or directly to the run() method.

The user interface can be either provided directly in the constructor, or it will be taken from the steering file. In the worst case, the fallback ConsoleInterface will be used.

The plugin manager, if not provided, the global plugin manager will be retrieved from the get_plugin_manager().

A simple example is provided here below:

Listing 29 Creation and execution of a MAFwApplication

import logging
from pathlib import Path

from mafw.runner import MAFwApplication

log = logging.getLogger(__name__)

# put here your steering file
steering_file = Path('path_to_my_steering_file.toml')

try:
    # create the app
    app = MAFwApplication(steering_file)

    # run it!
    app.run()

except Exception as e:
    log.error('An error occurred!')
    log.exception(e)

Changed in version 2.4: Added application-owned logical storage-service construction and external filter overlays.

Constructor parameters:

Parameters:

steering_file (Path | str, Optional) – The path to the steering file.
user_interface (UserInterfaceBase | type[UserInterfaceBase] | str, Optional) – The user interface to be used by the application.
filter_files – External filter TOML paths applied in supplied order.
plugin_manager (PluginManager, Optional) – The plugin manager.

_expand_processors_to_run(items: list[str], config: dict[str, Any], plugins: LoadedPlugins, seen: set[str] | None = None) → ProcessorList[source]

Constructs a processor.ProcessorList from a list of processor names or groups.

This method recursively expands a list of processor names or group definitions into a ProcessorList object, which can be executed. It handles three cases: direct processor class matches, group definitions in the configuration, and unknown entries.

Added in version v2.0.0.

Parameters:

items (list[str]) – A list of processor names or group identifiers to be expanded.
config (dict[str, Any]) – The configuration dictionary containing processor and group definitions.
plugins (LoadedPlugins) – The loaded plugins containing available processors and groups.
seen (set[str] | None) – A set of group names that have already been processed to detect cyclic definitions.

Returns:

A ProcessorList containing the expanded processors ready for execution.

Return type:

processor.ProcessorList

Raises:

UnknownProcessorGroup – If a group is defined without processors or if a cyclic group definition is detected.
UnknownProcessor – If a processor name is not recognized in the plugin library.

_resolve_processor(proc_name: str, plugins: LoadedPlugins) → ProcessorClassProtocol | None[source]

Resolve a parsed processor name to a processor class, or None when it may be a group.

The method wraps resolve() and adapts its outcome to the three-case dispatch of _expand_processors_to_run():

a unique short name or a known qualified name returns the resolved processor class;
an ambiguous short name lets AmbiguousProcessor propagate, since an under-specified name is a user error that must be surfaced rather than treated as a possible group;
a dotted qualified name that matches nothing raises UnknownProcessor naming the reference, rather than falling through to group resolution;
an undotted name that matches no processor returns None so the caller’s group resolution and final unknown-entry handling still apply.

Added in version v2.4.

Parameters:

proc_name (str) – The parsed name portion of a processors_to_run entry (replica already removed); it may be a dotted qualified name.
plugins (LoadedPlugins) – The loaded plugins providing the resolution lookup.

Returns:

The resolved processor class, or None when the name is not a processor.

Return type:

ProcessorClassProtocol | None

Raises:

AmbiguousProcessor – when the short name is shared by two or more processors.
UnknownProcessor – when a dotted qualified name matches no loaded processor.

get_user_interface(user_interface: str) → UserInterfaceBase[source]

Retrieves the user interface from the loaded plugins.

User interfaces are exposed via the plugin manager.

If the requested user_interface is not available, then the fallback console interface is used.

Parameters:: user_interface (str) – The name of the user interface to be used. Normally rich or console.

init(steering_file: Path | str) → None[source]

Initializes the application.

This method is normally automatically invoked by the class constructor. It can be called in a later moment to force the parsing of the provided steering file.

Parameters:: steering_file (Path | str) – The path to the steering file.

run(steering_file: Path | str | None = None) → ProcessorExitStatus[source]

Runs the application.

This method builds the ProcessorList with the processors listed in the steering file and launches its execution.

A steering file can be provided at this stage if it was not done before.

Changed in version 2.0: Refactor to accept replica names in the processor to run list.

Parameters:

steering_file (Path | str, Optional) – The steering file. Defaults to None.

Raises:

RunnerNotInitialized – if the application has not been initialized. Very likely a steering file was never provided.
UnknownProcessor – if a processor listed in the steering file is not available in the plugin library.

exit_status: the exit status of the application

filter_files: the name of the application instance

plugin_manager: the plugin manager of the application instance

storage_service: StorageService | None: The single service shared with this application’s processor tree, when configured.