acmecse.helpers.PluginManager.PluginManager

class documentation

class PluginManager:

Known subclasses: acmecse.runtime.PluginManager.PluginManager

PluginManager class.

This class manages the loading and registration of plugins.

Method	`__delattr__`	Delete the plugin by name.
Method	`__getattr__`	Get the instance or plugin by name.
Method	`callEndpoint`	Call a service plugin endpoint.
Method	`callEndpoints`	Call multiple service plugin endpoints. This is used to call the same endpoint on multiple plugins that match the given tag(s).
Method	`configurePlugins`	Configure the specified plugins.
Method	`dependencyGraph`	Get the dependency graph of the plugins and other classes.
Method	`dependencyGraphProvidedInstances`	Get the provided instances that are not plugins, and who is using them.
Method	`endpoints`	Get a list of all available endpoints provided by the plugin. This is used to discover the available endpoints and their metadata for a given plugin.
Method	`getPluginsByTag`	Get the names of the plugins that have the given tag.
Method	`has`	Check if a plugin instance with the given name exists.
Method	`isProvidedFunction`	Check if a provided instance with the given name is a function provided by the @provide decorator.
Method	`loadPlugins`	Load plugins from the specified directory.
Method	`pausePlugins`	Pause the specified plugins.
Method	`provide`	Provide a instance of any non-plugin class to be injected as a dependency. This can be used to provide instances of classes that are not plugins, but are required as dependencies by plugins or other classes.
Method	`resolvePlugins`	Resolve the dependencies of all plugins and other classes. This is called before starting the plugins to ensure that all dependencies are resolved and the plugin instances have all the required attributes injected.
Method	`restartPlugins`	Restart the specified plugins.
Method	`services`	Get a list of all available services provided by the plugins. This is used to discover the available services and their endpoints.
Method	`setupFinished`	Check if all required dependencies are resolved. This can be called after the resolvePlugins method to check if there are any unresolved dependencies that are required and not provided. This is a separate step because sometimes, we might want to resolve the plugins first and then check for missing dependencies, so that we can provide the missing dependencies before starting the plugins...
Method	`shutdownStarted`	Indicating that the system is shutting down. This can be used to perform any necessary cleanup.
Method	`startPlugins`	Start the specified plugins.
Method	`stopPlugins`	Shutdown the specified plugins. Plugins are stopped in reverse order of their priority.
Method	`unloadPlugins`	Unload a plugin by name. If running, stop it first.
Method	`unpausePlugins`	Unpause the specified plugins. Plugins are unpaused in reverse order of their priority.
Method	`unresolvePlugins`	Unresolve the plugins. This is called when plugins are unloaded to clean up the injected dependencies and set the plugin instances to None.
Method	`validatePlugins`	Validate the specified plugins.
Class Variable	`plugins`	Dictionary to hold the loaded plugins. The keys are the plugin names, the values are PluginInfo objects.
Class Variable	`unloadedPlugins`	List to hold the names of the plugins that were not loaded due to filtering.
Method	`_checkPluginTags`	Check if a plugin matches the specified tags and excluded tags.
Method	`_transition`	Transition the state of a plugin. This will happen in the order of the plugin priority, but can be reversed if necessary (e.g. for stopping plugins, which should happen in reverse order of starting).
Class Variable	`_pluginInstances`	Dictionary for the mapping between instance attribute names and the instances itself.
Class Variable	`_providedInstances`	Dictionary to hold instances that are extra provided class instances. The keys are the names of the instances, ie. the module name, the values are the instances themselves.
Class Variable	`_tagsPluginMap`	Dictionary to map tags to plugin names and instances for easier lookup by tag.

def __delattr__(self, name: str): ¶

Delete the plugin by name.

def __getattr__(self, name: str) -> Any: ¶

Get the instance or plugin by name.

Parameters
name:`str`	The name of the instance orplugin.

Returns
`Any`	The instance or plugin module, or None if not found.

def callEndpoint(self, pluginName: str, endpoint: str, *args: Any, timeout: float | None = None, **kwargs: Any) -> Any: ¶

Call a service plugin endpoint.

Parameters
pluginName:`str`	The name of the plugin to call. This is used to identify the plugin instance.
endpoint:`str`	The endpoint of the plugin to call. This is used to identify the method to call on the plugin instance. The endpoint must be defined in the plugin class using the `endpoint` decorator.
*args:`Any`	Positional arguments to pass to the endpoint method.
timeout:`float \| None`	Optional timeout value for the endpoint method call. If None, the default timeout value defined in the plugin class or method is used.
**kwargs:`Any`	Keyword arguments to pass to the endpoint method.

Returns
`Any`	The result of the endpoint method call.

Raises
`PluginNotFoundError`	If no plugin with the given name is found.
`EndpointNotFoundError`	If no plugin with the given endpoint is found.
`EndpointTimeoutError`	If the endpoint method call times out.

def callEndpoints(self, endpoint: str, *args: Any, tag: str | None | list[str] = None, timeout: float | None = None, **kwargs: Any) -> list[tuple[Any, str, dict[str, Any]]]: ¶

Call multiple service plugin endpoints. This is used to call the same endpoint on multiple plugins that match the given tag(s).

Parameters
endpoint:`str`	The endpoint of the plugin to call. This is used to identify the method to call on the plugin instance. The endpoint must be defined in the plugin class using the `endpoint` decorator.
*args:`Any`	Positional arguments to pass to the endpoint method.
tag:`str \| None \| list[str]`	The optional tag or tags of the plugin to call. This is used to identify the plugin to call. If multiple plugins with the same tag are found, all of them are called in order of their priority.
timeout:`float \| None`	Optional timeout value for the endpoint method call. If None, the default timeout value defined in the plugin class or method is used.
**kwargs:`Any`	Keyword arguments to pass to the endpoint method.

Returns
`list[tuple[Any, str, dict[str, Any]]]`	The result of the endpoint method call. This is returned as a list of tuples of the form (result, pluginName) where result is the return value of the endpoint method and pluginName is the name of the plugin that was called.

Raises
`PluginNotFoundError`	If no plugin with the given tag and endpoint is found.

Configure the specified plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to configure. Configure all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.
*args:`Any`	Positional arguments to pass to the configure method.
**kwargs:`Any`	Keyword arguments to pass to the configure method.

Raises
`KeyError`	If a specified plugin is not found.

def dependencyGraph(self, name: str | None = None) -> dict[str, list[Dependency]]: ¶

Get the dependency graph of the plugins and other classes.

Parameters
name:`str \| None`	Optional name of the plugin to get the dependency graph for. If None, returns the entire graph.

Returns
`dict[str, list[Dependency]]`	The dependency graph as a dictionary where the keys are the class names and the values are lists of dependencies (module name, optional instance name).

def dependencyGraphProvidedInstances(self) -> dict[tuple[str, bool], list[tuple[str, str, str]]]: ¶

Get the provided instances that are not plugins, and who is using them.

Returns
`dict[tuple[str, bool], list[tuple[str, str, str]]]`	A dictionary where the keys are tuples of the form (provided instance name, isFunction flag) and the values are a list of tuples of the form (dependent module name, class name, attribute name) that are using the provided instance as a dependency.

def endpoints(self, pluginName: str) -> list[tuple[str, inspect.Signature]]: ¶

Get a list of all available endpoints provided by the plugin. This is used to discover the available endpoints and their metadata for a given plugin.

Parameters
pluginName:`str`	The name of the plugin to get the endpoints for.

Returns
`list[tuple[str, inspect.Signature]]`	A list of tuples of the form (endpoint, signature) where endpoint is the name of the endpoint and signature a Signature object from the inspect module.

Raises
`PluginNotFoundError`	If no plugin with the given name is found.

def getPluginsByTag(self, tags: str | list[str], byPriority: bool = False) -> list[tuple[str, Any]]: ¶

Get the names of the plugins that have the given tag.

Parameters
tags:`str \| list[str]`	The tag or tags to search for.
byPriority:`bool`	Whether to sort the plugins by their priority. If True, the plugins are returned sorted by their priority, with the highest priority first (0 = highest priority).

Returns
`list[tuple[str, Any]]`	A list of plugin names that have the given tag, sorted by priority if requested.

def has(self, instanceName: str) -> bool: ¶

Check if a plugin instance with the given name exists.

Parameters
instanceName:`str`	The name of the plugin instance to check.

Returns
`bool`	True if a plugin instance with the given name exists, False otherwise.

def isProvidedFunction(self, name: str) -> bool: ¶

Check if a provided instance with the given name is a function provided by the @provide decorator.

Parameters
name:`str`	The name of the provided instance to check.

Returns
`bool`	True if a provided instance with the given name is a function provided by the @provide decorator, False otherwise.

def loadPlugins(self, directory: str, packagePath: str, pluginFilter: Callable[[str], bool] | None = None, replace: bool = False, timeout: float | None = 1.0, *args: Any, **kwargs: Any) -> dict[str, PluginInfo]: ¶

Load plugins from the specified directory.

Plugins are initialized after loading according to their priority.

Parameters
directory:`str`	The directory from which to load plugins.
packagePath:`str`	The package path to use for the plugins.
pluginFilter:`Callable[[str], bool] \| None`	A callback function to filter plugins. The function should take a plugin name as input and return True if the plugin should be loaded, False otherwise.
replace:`bool`	Whether to replace already loaded plugins.
timeout:`float \| None`	Optional timeout for plugins' methods.
*args:`Any`	Positional arguments to pass to the plugin init methods.
**kwargs:`Any`	Keyword arguments to pass to the plugin init methods.

Returns
`dict[str, PluginInfo]`	A dictionary of the newly loaded plugins.

Raises
`NotADirectoryError`	If the directory does not exist.
`KeyError`	If a plugin is already loaded.
`PluginConfigurationError`	If the plugin is not valid.

Pause the specified plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to pause. Pause all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.

Raises
`KeyError`	If a specified plugin is not found.

def provide(self, moduleName: str, instance: Any): ¶

Provide a instance of any non-plugin class to be injected as a dependency. This can be used to provide instances of classes that are not plugins, but are required as dependencies by plugins or other classes.

The provided instances are injected into the dependent classes during the dependency resolution phase, just like plugin instances.

The provided instances are not managed by the plugin manager, so they are not started, stopped, etc. They are simply injected as attributes into the dependent classes.

Parameters
moduleName:`str`	The name of the module that provides the instance. This is used to identify the instance in the dependency graph and to inject it into the dependent classes.
instance:`Any`	The instance to provide. This can be any instance of any class, as long as it is not a plugin class (i.e. it does not have the @pluginClass decorator).

def resolvePlugins(self, tags: str | list[str] | None = None, excludedTags: str | list[str] | None = None): ¶

Resolve the dependencies of all plugins and other classes. This is called before starting the plugins to ensure that all dependencies are resolved and the plugin instances have all the required attributes injected.

Parameters
tags:`str \| list[str] \| None`	The tags of the plugins to match for dependency resolution. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude from dependency resolution. Exclude none

Raises
`DependencyError`	If a required dependency cannot be resolved.

Restart the specified plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to restart. Restart all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.

Raises
`KeyError`	If a specified plugin is not found.

def services(self, tag: str | list[str] = None) -> list[tuple[str, list[str], dict[str, Any]]]: ¶

Get a list of all available services provided by the plugins. This is used to discover the available services and their endpoints.

Parameters
tag:`str \| list[str]`	The tag of the plugin to get the services for. This is used to identify the plugin to get the services for. If multiple plugins with the same tag are found, all of them are returned in order of their priority. If None, all plugins are returned.

Returns
`list[tuple[str, list[str], dict[str, Any]]]`	A list of tuples of the form (pluginName, tags, serviceMetadata) where pluginName is the name of the plugin that provides the service, tags is a list of tags associated with the plugin, and serviceMetadata is a dictionary of metadata about the service (e.g. description, parameters, etc.).

def setupFinished(self): ¶

Check if all required dependencies are resolved. This can be called after the resolvePlugins method to check if there are any unresolved dependencies that are required and not provided. This is a separate step because sometimes, we might want to resolve the plugins first and then check for missing dependencies, so that we can provide the missing dependencies before starting the plugins. For example, we might want to provide some extra instances that are not plugins, but are required by some plugins or other classes. In this case, we can call resolvePlugins first, then setupFinished to see if there are any missing dependencies, then provide the missing instances, and then start the plugins.

def shutdownStarted(self) -> bool: ¶

Indicating that the system is shutting down. This can be used to perform any necessary cleanup.

However, this does not stop or unload the plugins, since we might want to perform some other cleanup.

Returns
`bool`	Always returns True.

Start the specified plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to start. Start all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.

Raises
`KeyError`	If a specified plugin is not found.

Shutdown the specified plugins. Plugins are stopped in reverse order of their priority.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to shutdown. Shutdown all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.

Raises
`KeyError`	If a specified plugin is not found.

def unloadPlugins(self, pluginNames: list[str] | str | None = None): ¶

Unload a plugin by name. If running, stop it first.

Parameters
pluginNames:`list[str] \| str \| None`	The name(s) of the plugin(s) to remove.

Raises
`KeyError`	If the plugin is not found.

Unpause the specified plugins. Plugins are unpaused in reverse order of their priority.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to unpause. Unpause all if None.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.

Raises
`KeyError`	If a specified plugin is not found.

def unresolvePlugins(self): ¶

Unresolve the plugins. This is called when plugins are unloaded to clean up the injected dependencies and set the plugin instances to None.

Validate the specified plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to validate.
tags:`str \| list[str] \| None`	The tags of the plugins to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags of the plugins to exclude. Exclude none if None.
*args:`Any`	Positional arguments to pass to the validate method.
**kwargs:`Any`	Keyword arguments to pass to the validate method.

Raises
`KeyError`	If a specified plugin is not found.

plugins: dict[str, PluginInfo] = ¶

Dictionary to hold the loaded plugins. The keys are the plugin names, the values are PluginInfo objects.

unloadedPlugins: list[str] = ¶

List to hold the names of the plugins that were not loaded due to filtering.

def _checkPluginTags(self, pluginName: str, tags: str | list[str] | None = None, excludedTags: str | list[str] | None = None) -> bool: ¶

Check if a plugin matches the specified tags and excluded tags.

Parameters
pluginName:`str`	The name of the plugin to check.
tags:`str \| list[str] \| None`	The tags to match. Match all if None.
excludedTags:`str \| list[str] \| None`	The tags to exclude. Exclude none if None.

Returns
`bool`	True if the plugin matches the tags and excluded tags, False otherwise.

Transition the state of a plugin. This will happen in the order of the plugin priority, but can be reversed if necessary (e.g. for stopping plugins, which should happen in reverse order of starting).

Plugins can be filtered by name, pattern and/or tags. If no filters are provided, the action will be applied to all plugins.

Parameters
pluginNames:`str \| list[str] \| None`	The name(s) of the plugin(s) to transition. If None, all plugins are transitioned. If a string is provided, it is treated as a pattern to match plugin names.
action:`Callable[[PluginInfo], None]`	The action to perform during the state transition.
tags:`str \| list[str] \| None`	The tags to match for the plugins to transition. If None, all plugins are transitioned.
excludedTags:`str \| list[str] \| None`	The tags to exclude from the plugins to transition. If None, no plugins are excluded.
reverse:`bool`	Whether to process the plugins in reverse order.

_pluginInstances: dict[str, Any] = ¶

Dictionary for the mapping between instance attribute names and the instances itself.

_providedInstances: dict[str, Any] = ¶

Dictionary to hold instances that are extra provided class instances. The keys are the names of the instances, ie. the module name, the values are the instances themselves.

_tagsPluginMap: dict[str, list[tuple[str, Any]]] = ¶

Dictionary to map tags to plugin names and instances for easier lookup by tag.