@ -4,6 +4,7 @@
from __future__ import annotations
from __future__ import annotations
import dataclasses
import os
import os
from ansible import context
from ansible import context
@ -14,6 +15,7 @@ from ansible.module_utils.common.text.converters import to_native, to_bytes
from ansible . plugins import loader
from ansible . plugins import loader
from ansible . utils . display import Display
from ansible . utils . display import Display
from ansible . utils . collection_loader . _collection_finder import _get_collection_path
from ansible . utils . collection_loader . _collection_finder import _get_collection_path
from ansible . _internal . _templating . _jinja_plugins import get_jinja_builtin_plugin_descriptions
display = Display ( )
display = Display ( )
@ -25,6 +27,20 @@ IGNORE = {
}
}
@dataclasses.dataclass ( kw_only = True , frozen = True , slots = True )
class _PluginDocMetadata :
""" Information about a plugin. """
name : str
""" The fully qualified name of the plugin. """
path : bytes | None = None
""" The path to the plugin file, or None if not available. """
plugin_obj : object | None = None
""" The loaded plugin object, or None if not loaded. """
jinja_builtin_short_description : str | None = None
""" The short description of the plugin if it is a Jinja builtin, otherwise None. """
def get_composite_name ( collection , name , path , depth ) :
def get_composite_name ( collection , name , path , depth ) :
resolved_collection = collection
resolved_collection = collection
if ' . ' not in name :
if ' . ' not in name :
@ -116,21 +132,37 @@ def _list_j2_plugins_from_file(collection, plugin_path, ptype, plugin_name):
return file_plugins
return file_plugins
def list_collection_plugins ( ptype , collections , search_paths = None ) :
def list_collection_plugins ( ptype : str , collections : dict [ str , bytes ] , search_paths : list [ str ] | None = None ) - > dict [ str , tuple [ bytes , object | None ] ] :
# Kept for backwards compatibility.
return {
name : ( info . path , info . plugin_obj )
for name , info in _list_collection_plugins_with_info ( ptype , collections ) . items ( )
}
def _list_collection_plugins_with_info (
ptype : str ,
collections : dict [ str , bytes ] ,
) - > dict [ str , _PluginDocMetadata ] :
# TODO: update to use importlib.resources
# TODO: update to use importlib.resources
# starts at {plugin_name: filepath, ...}, but changes at the end
plugins = { }
try :
try :
ploader = getattr ( loader , ' {0} _loader ' . format ( ptype ) )
ploader = getattr ( loader , ' {0} _loader ' . format ( ptype ) )
except AttributeError :
except AttributeError :
raise AnsibleError ( f " Cannot list plugins, incorrect plugin type { ptype !r} supplied. " ) from None
raise AnsibleError ( f " Cannot list plugins, incorrect plugin type { ptype !r} supplied. " ) from None
builtin_jinja_plugins = { }
plugin_paths = { }
# get plugins for each collection
# get plugins for each collection
for collection in collections . keys ( ) :
for collection , path in collections . item s( ) :
if collection == ' ansible.builtin ' :
if collection == ' ansible.builtin ' :
# dirs from ansible install, but not configured paths
# dirs from ansible install, but not configured paths
dirs = [ d . path for d in ploader . _get_paths_with_context ( ) if d . internal ]
dirs = [ d . path for d in ploader . _get_paths_with_context ( ) if d . internal ]
if ptype in ( ' filter ' , ' test ' ) :
builtin_jinja_plugins = get_jinja_builtin_plugin_descriptions ( ptype )
elif collection == ' ansible.legacy ' :
elif collection == ' ansible.legacy ' :
# configured paths + search paths (should include basedirs/-M)
# configured paths + search paths (should include basedirs/-M)
dirs = [ d . path for d in ploader . _get_paths_with_context ( ) if not d . internal ]
dirs = [ d . path for d in ploader . _get_paths_with_context ( ) if not d . internal ]
@ -139,7 +171,7 @@ def list_collection_plugins(ptype, collections, search_paths=None):
else :
else :
# search path in this case is for locating collection itselfA
# search path in this case is for locating collection itselfA
b_ptype = to_bytes ( C . COLLECTION_PTYPE_COMPAT . get ( ptype , ptype ) )
b_ptype = to_bytes ( C . COLLECTION_PTYPE_COMPAT . get ( ptype , ptype ) )
dirs = [ to_native ( os . path . join ( collections[ collection ] , b ' plugins ' , b_ptype ) ) ]
dirs = [ to_native ( os . path . join ( path , b ' plugins ' , b_ptype ) ) ]
# acr = AnsibleCollectionRef.try_parse_fqcr(collection, ptype)
# acr = AnsibleCollectionRef.try_parse_fqcr(collection, ptype)
# if acr:
# if acr:
# dirs = acr.subdirs
# dirs = acr.subdirs
@ -147,30 +179,51 @@ def list_collection_plugins(ptype, collections, search_paths=None):
# raise Exception('bad acr for %s, %s' % (collection, ptype))
# raise Exception('bad acr for %s, %s' % (collection, ptype))
plugin s. update ( _list_plugins_from_paths ( ptype , dirs , collection ) )
plugin _path s. update ( _list_plugins_from_paths ( ptype , dirs , collection ) )
# return plugin and it's class object, None for those not verifiable or failing
plugins = { }
if ptype in ( ' module ' , ) :
if ptype in ( ' module ' , ) :
# no 'invalid' tests for modules
# no 'invalid' tests for modules
for plugin in plugins . key s( ) :
for plugin , plugin_path in plugin_paths . item s( ) :
plugins [ plugin ] = ( plugins [ plugin ] , None )
plugins [ plugin ] = _PluginDocMetadata ( name = plugin , path = plugin_path )
else :
else :
# detect invalid plugin candidates AND add loaded object to return data
# detect invalid plugin candidates AND add loaded object to return data
for plugin in list ( plugins . keys ( ) ) :
for plugin , plugin_path in plugin_paths . items ( ) :
pobj = None
pobj = None
try :
try :
pobj = ploader . get ( plugin , class_only = True )
pobj = ploader . get ( plugin , class_only = True )
except Exception as e :
except Exception as e :
display . vvv ( " The ' {0} ' {1} plugin could not be loaded from ' {2} ' : {3} " . format ( plugin , ptype , plugin s[ plugin ] , to_native ( e ) ) )
display . vvv ( " The ' {0} ' {1} plugin could not be loaded from ' {2} ' : {3} " . format ( plugin , ptype , plugin _path , to_native ( e ) ) )
# sets final {plugin_name: (filepath, class|NONE if not loaded), ...}
plugins [ plugin ] = _PluginDocMetadata (
plugins [ plugin ] = ( plugins [ plugin ] , pobj )
name = plugin ,
path = plugin_path ,
plugin_obj = pobj ,
jinja_builtin_short_description = builtin_jinja_plugins . get ( plugin ) ,
)
# Add in any builtin Jinja2 plugins that have not been shadowed in Ansible.
plugins . update (
( plugin_name , _PluginDocMetadata ( name = plugin_name , jinja_builtin_short_description = plugin_description ) )
for plugin_name , plugin_description in builtin_jinja_plugins . items ( ) if plugin_name not in plugins
)
# {plugin_name: (filepath, class), ...}
return plugins
return plugins
def list_plugins ( ptype , collections = None , search_paths = None ) :
def list_plugins ( ptype : str , collections : list [ str ] | None = None , search_paths : list [ str ] | None = None ) - > dict [ str , tuple [ bytes , object | None ] ] :
# Kept for backwards compatibility.
return {
name : ( info . path , info . plugin_obj )
for name , info in _list_plugins_with_info ( ptype , collections , search_paths ) . items ( )
}
def _list_plugins_with_info (
ptype : str ,
collections : list [ str ] = None ,
search_paths : list [ str ] | None = None ,
) - > dict [ str , _PluginDocMetadata ] :
if isinstance ( collections , str ) :
if isinstance ( collections , str ) :
collections = [ collections ]
collections = [ collections ]
@ -195,7 +248,7 @@ def list_plugins(ptype, collections=None, search_paths=None):
raise AnsibleError ( f " Cannot use supplied collection { collection !r} . " ) from ex
raise AnsibleError ( f " Cannot use supplied collection { collection !r} . " ) from ex
if plugin_collections :
if plugin_collections :
plugins . update ( list_collection_plugins( ptype , plugin_collections ) )
plugins . update ( _ list_collection_plugins_with_info ( ptype , plugin_collections ) )
return plugins
return plugins