Classes
class	PluginBase

Functions
str	_get_plugin_path (str plugin, GulpPluginType plugin_type=GulpPluginType.INGESTION)

list[str]	get_plugin_tags (str plugin)

list[dict]	list_plugins ()

PluginBase	load_plugin (str plugin, GulpPluginType plugin_type=GulpPluginType.INGESTION, bool ignore_cache=False, **kwargs)

None	plugin_cache_add (ModuleType m, str name)

None	plugin_cache_clear ()

ModuleType	plugin_cache_get (str plugin)

None	plugin_cache_remove (str plugin)

None	unload_plugin (PluginBase mod)

Variables
dict	_cache = {}

Detailed Description

Gulp plugin base class and plugin utilities.

Function Documentation

◆ _get_plugin_path()

str _get_plugin_path	(	str	plugin,
		GulpPluginType	plugin_type = GulpPluginType.INGESTION )

protected

Definition at line 1017 of file plugin.py.

) -> str:
 
    # try plain .py first
    # TODO: on license manager, disable plain .py load (only encrypted pyc)
    # get path according to plugin type
    path_plugins = config.path_plugins(plugin_type)
 
    ppy = muty.file.safe_path_join(path_plugins, plugin + ".py")
    if not muty.file.exists(ppy):
        # try pyc
        ppyc = muty.file.safe_path_join(path_plugins, plugin + ".pyc")
        if not muty.file.exists(ppyc):
            raise ObjectNotFound(
                "plugin %s not found (tried %s, %s)" % (plugin, ppy, ppyc)
            )
        return ppyc
    return ppy
 
 

Here is the caller graph for this function:

◆ get_plugin_tags()

list[str] get_plugin_tags ( str plugin )

Get the tags for a given (ingestion) plugin.

Args:
    plugin (str): The name of the plugin to get the tags for.
Returns:
    list[str]: The tags for the given plugin.

Definition at line 1131 of file plugin.py.

async def get_plugin_tags(plugin: str) -> list[str]:
    """
    Get the tags for a given (ingestion) plugin.
 
    Args:
        plugin (str): The name of the plugin to get the tags for.
    Returns:
        list[str]: The tags for the given plugin.
    """
    p = load_plugin(plugin)
    tags = p.tags()
    return tags
 
 

Here is the call graph for this function:

◆ list_plugins()

list[dict] list_plugins ( )

List all available plugins.

Returns:
    list[dict]: The list of available plugins.

Definition at line 1095 of file plugin.py.

async def list_plugins() -> list[dict]:
    """
    List all available plugins.
 
    Returns:
        list[dict]: The list of available plugins.
    """
    path_plugins = config.path_plugins()
    files = await muty.file.list_directory_async(path_plugins, "*.py*", recursive=True)
    l = []
    for f in files:
        if "__init__" not in f and "__pycache__" not in f:
            try:
                p = load_plugin(f)
                n = {
                    "display_name": p.name(),
                    "type": str(p.type()),
                    "desc": p.desc(),
                    "filename": os.path.basename(p.path),
                    "internal": p.internal(),
                    "options": [o.to_dict() for o in p.options()],
                    "depends_on": p.depends_on(),
                    "tags": p.tags(),
                    "event_type_field": p.event_type_field(),
                    "version": p.version(),
                }
                l.append(n)
                unload_plugin(p)
            except Exception as ex:
                logger().exception(ex)
                logger().error("could not load plugin %s" % (f))
                continue
 
    return l
 
 

Here is the call graph for this function:

◆ load_plugin()

PluginBase load_plugin	(	str	plugin,
		GulpPluginType	plugin_type = GulpPluginType.INGESTION,
		bool	ignore_cache = False,
		**	kwargs )

Load a plugin from a given path or from the default plugin path.

Args:
    plugin (str): The name or path of the plugin to load.
    plugin_type (GulpPluginType, optional): The type of the plugin to load. Defaults to GulpPluginType.INGESTION.
        this is ignored if the plugin is an absolute path or if "plugin_cache" is enabled and the plugin is already cached.
    ignore_cache (bool, optional): Whether to ignore the plugin cache. Defaults to False.
    **kwargs (dict, optional): Additional keyword arguments.
Returns:
    PluginBase: The loaded plugin.

Raises:
    Exception: If the plugin could not be loaded.

Definition at line 1038 of file plugin.py.

) -> PluginBase:
    """
    Load a plugin from a given path or from the default plugin path.
 
    Args:
        plugin (str): The name or path of the plugin to load.
        plugin_type (GulpPluginType, optional): The type of the plugin to load. Defaults to GulpPluginType.INGESTION.
            this is ignored if the plugin is an absolute path or if "plugin_cache" is enabled and the plugin is already cached.
        ignore_cache (bool, optional): Whether to ignore the plugin cache. Defaults to False.
        **kwargs (dict, optional): Additional keyword arguments.
    Returns:
        PluginBase: The loaded plugin.
 
    Raises:
        Exception: If the plugin could not be loaded.
    """
    logger().debug("load_plugin %s ..." % (plugin))
 
    if not '/' in plugin and (plugin.lower().endswith(".py") or plugin.lower().endswith(".pyc")):
        # remove extension
        plugin = plugin.rsplit(".", 1)[0]
 
    m = plugin_cache_get(plugin)
    if ignore_cache:
        logger().debug("ignoring cache for plugin %s" % (plugin))
        m = None
 
    if m is not None:
        return m.Plugin(plugin, **kwargs)
 
    if "/" in plugin:
        # plugins is an absolute path
        path = muty.file.abspath(plugin)
    else:
        # uses plugin_type to load from the correct subfolder
        path = _get_plugin_path(plugin, plugin_type=plugin_type)
 
    try:
        m: PluginBase = muty.dynload.load_dynamic_module_from_file(path)
    except Exception as ex:
        raise Exception(
            "load_dynamic_module_from_file() failed, could not load plugin %s !"
            % (path)
        ) from ex
 
    mod: PluginBase = m.Plugin(path, **kwargs)
    logger().debug("loaded plugin: %s" % (mod.name()))
    plugin_cache_add(m, plugin)
 
    return mod
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ plugin_cache_add()

None plugin_cache_add	(	ModuleType	m,
		str	name )

Add a plugin to the process's own plugin cache.

Args:
    m (ModuleType): The plugin module to add to the cache.
    name (str): The name/path of the plugin.

Returns:
    None

Definition at line 1206 of file plugin.py.

def plugin_cache_add(m: ModuleType, name: str) -> None:
    """
    Add a plugin to the process's own plugin cache.
 
    Args:
        m (ModuleType): The plugin module to add to the cache.
        name (str): The name/path of the plugin.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    mm = _cache.get(name, None)
    if mm is None:
        logger().debug("adding plugin %s (%s) to cache" % (name, m))
        _cache[name] = m
 
 

Here is the caller graph for this function:

◆ plugin_cache_clear()

None plugin_cache_clear ( )

Clear the process's own plugin cache.

Returns:
    None

Definition at line 1170 of file plugin.py.

def plugin_cache_clear() -> None:
    """
    Clear the process's own plugin cache.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    _cache = {}
 
 

◆ plugin_cache_get()

ModuleType plugin_cache_get ( str plugin )

Retrieve a plugin from the process's own plugin cache.

Args:
    plugin (str): The name/path of the plugin to retrieve.

Returns:
    ModuleType: The plugin module if found in the cache, otherwise None.

Definition at line 1227 of file plugin.py.

def plugin_cache_get(plugin: str) -> ModuleType:
    """
    Retrieve a plugin from the process's own plugin cache.
 
    Args:
        plugin (str): The name/path of the plugin to retrieve.
 
    Returns:
        ModuleType: The plugin module if found in the cache, otherwise None.
    """
    global _cache
    if not config.plugin_cache_enabled():
        return None
 
    p = _cache.get(plugin, None)
    if p is not None:
        logger().debug("found plugin %s in cache" % (plugin))
    else:
        logger().warning("plugin %s not found in cache" % (plugin))
    return p

Here is the caller graph for this function:

◆ plugin_cache_remove()

None plugin_cache_remove ( str plugin )

Remove a plugin from the process's own plugin cache.

Args:
    plugin (str): The name/path of the plugin to remove from the cache.

Returns:
    None

Definition at line 1184 of file plugin.py.

def plugin_cache_remove(plugin: str) -> None:
    """
    Remove a plugin from the process's own plugin cache.
 
    Args:
        plugin (str): The name/path of the plugin to remove from the cache.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    if plugin in _cache:
        logger().debug("removing plugin %s from cache" % (plugin))
 
        # cleanup module and delete
        m = _cache[plugin]
        del _cache[plugin]
 
 

◆ unload_plugin()

None unload_plugin ( PluginBase mod )

Unloads a plugin module by calling its `unload` method and deletes the module object

NOTE: mod is **no more valid** after this function returns.

Args:
    mod (PluginBase): The plugin module to unload.
    run_gc (bool): if set, garbage collector is called after unloading the module. Defaults to True.

Returns:
    None

Definition at line 1145 of file plugin.py.

def unload_plugin(mod: PluginBase) -> None:
    """
    Unloads a plugin module by calling its `unload` method and deletes the module object
 
    NOTE: mod is **no more valid** after this function returns.
 
    Args:
        mod (PluginBase): The plugin module to unload.
        run_gc (bool): if set, garbage collector is called after unloading the module. Defaults to True.
 
    Returns:
        None
    """
    if config.plugin_cache_enabled():
        return
 
    if mod is not None:
        # delete from cache if any
        # plugin_cache_delete(mod)
 
        logger().debug("unloading plugin: %s" % (mod.name()))
        mod.cleanup()
        del mod
 
 

Here is the caller graph for this function:

Variable Documentation

◆ _cache

dict _cache = {}

protected

Definition at line 43 of file plugin.py.

Classes

Functions

Variables

Detailed Description

Function Documentation

◆ _get_plugin_path()

◆ get_plugin_tags()

◆ list_plugins()

◆ load_plugin()

◆ plugin_cache_add()

◆ plugin_cache_clear()

◆ plugin_cache_get()

◆ plugin_cache_remove()

◆ unload_plugin()

Variable Documentation

◆ _cache