"""
datasources - Provides introspection functionality for datasources in insights-core
This module provides classes to provide introspection for each of the datasource
definition classes in the :py:mod`insights.specs` package.
"""
import inspect
import insights.core.spec_factory
import insights.specs
from insights.specs.default import DefaultSpecs as InDefaultSpecs
from insights.specs.insights_archive import InsightsArchiveSpecs as InArchiveSpecs
from insights.specs.sos_archive import SosSpecs as InSosSpecs
from . import AutologyError
SIMPLE_COMMAND_TYPE = 'simple_command'
""" str: Literal constant for a simple_comment Spec object """
SIMPLE_FILE_TYPE = 'simple_file'
""" str: Literal constant for a simple_file Spec object """
GLOB_FILE_TYPE = 'glob_file'
""" str: Literal constant for a glob_file Spec object """
FOREACH_EXECUTE_TYPE = 'foreach_execute'
""" str: Literal constant for a foreach_execute Spec object """
CONTAINER_EXECUTE_TYPE = 'container_execute'
""" str: Literal constant for a container_execute Spec object """
LISTDIR_TYPE = 'listdir'
""" str: Literal constant for a listdir Spec object """
LIST_TYPE = 'list'
""" str: Literal constant for a list Spec object """
STRING_TYPE = 'string'
""" str: Literal constant for a string Spec object """
NONE_TYPE = 'none'
""" str: Literal constant for a Spec object with None type """
FUNCTION_TYPE = 'function'
""" str: Literal constant for a function Spec object """
FIRST_FILE_TYPE = 'first_file'
""" str: Literal constant for a first_file Spec object """
FOREACH_COLLECT_TYPE = 'foreach_collect'
""" str: Literal constant for a foreach_collect Spec object """
CONTAINER_COLLECT_TYPE = 'container_collect'
""" str: Literal constant for a container_collect Spec object """
FIRST_OF_TYPE = 'first_of'
""" str: Literal constant for a first_of Spec object """
COMMAND_WITH_ARGS_TYPE = 'command_with_args'
""" str: Literal constant for a command_with_args Spec object """
HEAD_TYPE = 'head'
""" str: Literal constant used for a head Spec object """
UNKNOWN_TYPE = 'unknown'
""" str: Literal constant for a Spec object with unknown type """
ANONYMOUS_SPEC_NAME = 'anonymous'
""" str: Literal constant used for Specs with no ``name`` attribute """
[docs]
def is_simple_command(m_obj):
""" bool: True if broker object is a simple_command object """
return isinstance(m_obj, insights.core.spec_factory.simple_command)
[docs]
def is_simple_file(m_obj):
""" bool: True if broker object is a simple_file object """
return isinstance(m_obj, insights.core.spec_factory.simple_file)
[docs]
def is_glob_file(m_obj):
""" bool: True if broker object is a glob_file object """
return isinstance(m_obj, insights.core.spec_factory.glob_file)
[docs]
def is_foreach_execute(m_obj):
""" bool: True if broker object is a foreach_execute object """
return isinstance(m_obj, insights.core.spec_factory.foreach_execute)
[docs]
def is_container_execute(m_obj):
""" bool: True if broker object is a is_container_execute object """
return isinstance(m_obj, insights.core.spec_factory.container_execute)
[docs]
def is_first_file(m_obj):
""" bool: True if broker object is a first_file object """
return isinstance(m_obj, insights.core.spec_factory.first_file)
[docs]
def is_first_of(m_obj):
""" bool: True if broker object is a is_first_of object """
return isinstance(m_obj, insights.core.spec_factory.first_of)
[docs]
def is_foreach_collect(m_obj):
""" bool: True if broker object is a is_foreach_collect object """
return isinstance(m_obj, insights.core.spec_factory.foreach_collect)
[docs]
def is_container_collect(m_obj):
""" bool: True if broker object is a is_container_collect object """
return isinstance(m_obj, insights.core.spec_factory.container_collect)
[docs]
def is_listdir(m_obj):
""" bool: True if broker object is a is_listdir object """
return isinstance(m_obj, insights.core.spec_factory.listdir)
[docs]
def is_command_with_args(m_obj):
""" bool: True if broker object is a is_command_with_args object """
return isinstance(m_obj, insights.core.spec_factory.command_with_args)
[docs]
def is_head(m_obj):
""" bool: True if object is a head object """
return isinstance(m_obj, insights.core.spec_factory.head)
[docs]
def is_function(m_obj):
""" bool: True if object is a function object """
return inspect.isfunction(m_obj)
[docs]
class Spec(dict):
"""
Class to identify and describe datasources and related objects
The normal way to create one of these objects is to use the factory
method :py:meth:`from_object` and provide as input one of the datasource
objects from :py:mod:`insights.core.spec_factory`. This class is implemented
as a dictionary and all attributes of the object are stored as dictionary keys.
A ``repr`` string must be included in ``kwargs`` to __init__ and it will be
removed and stored in the ``repr_str`` attribute. If a ``name`` attributed
is not provided, a default name of ``anonymous`` will be used. This is a special
name recognized when formatting Spec output to be used when specs have other
specs as providers/dependents.
Attributes:
repr_str (str): String to be used to implement the __repr__ method for this object
Supports jinja2 formatting using dictionary attributes of this object
self (dict): All attributes of this object are included in the dictionary except
the repr string.
Raises:
AutologyError: Raised if a ``repr`` string is not provided or if an unsupported
object type is passed to the constructor or to the factory method.
"""
def __init__(self, **kwargs):
super(Spec, self).__init__(kwargs)
try:
self.repr_str = self.pop('repr')
except KeyError:
raise AutologyError('A repr items must be supplied for the spec name: {}'.format(self.get('name', 'no name')))
if 'name' not in self:
self['name'] = ANONYMOUS_SPEC_NAME
def __repr__(self):
try:
formatted_str = self.repr_str.format(**self)
except KeyError:
if '{name} =' in self.repr_str:
_, r = self.repr_str.split('{name} =')
fixed = r.replace('{', '{{').replace('}', '}}')
formatted_str = (''.join(['{name} =', fixed])).format(**self)
else:
formatted_str = self.repr_str.replace('{', '{{').replace('}', '}}').format(**self)
return formatted_str
[docs]
@classmethod
def from_object(cls, m_type, m_name=ANONYMOUS_SPEC_NAME):
"""
Factory method to create Spec objects
This method evaluates the m_type object type and extract the Spec
attributes based on that type.
Attributes:
m_type (obj): One of the datasource objects from :py:mod:`insights.core.spec_factory`.
m_name (str): Name of the datasource object, if not provided ``ANONYMOUS_SPEC_NAME``
Raises:
AutologyError: Raises this error if it cannot determine the object type
Returns:
Spec: Returns an object of type `Spec`
"""
m_members = inspect.getmembers(m_type)
m_spec = {'name': m_name, 'type': m_type}
m_spec['kind'] = next((v for k, v in m_members if k == "kind"), None)
m_spec['context'] = next((v for k, v in m_members if k == "context"), None)
m_spec['multi_output'] = next((v for k, v in m_members if k == "multi_output"), None)
if is_simple_command(m_type):
m_spec['type_name'] = SIMPLE_COMMAND_TYPE
m_spec['cmd'] = next((v for k, v in m_members if k == "cmd"), None)
m_spec['repr'] = 'simple_command("{cmd}")'
elif is_simple_file(m_type):
m_spec['type_name'] = SIMPLE_FILE_TYPE
m_spec['path'] = next((v for k, v in m_members if k == "path"), None)
m_spec['repr'] = 'simple_file("{path}")'
elif is_glob_file(m_type):
m_spec['type_name'] = GLOB_FILE_TYPE
m_spec['patterns'] = next((v for k, v in m_members if k == "patterns"), None)
m_spec['repr'] = 'glob_file({patterns})'
elif is_first_file(m_type):
m_spec['type_name'] = FIRST_FILE_TYPE
m_spec['paths'] = next((v for k, v in m_members if k == "paths"), None)
m_spec['repr'] = 'first_file({paths})'
elif is_listdir(m_type):
m_spec['type_name'] = LISTDIR_TYPE
m_spec['path'] = next((v for k, v in m_members if k == 'path'), None)
m_spec['repr'] = 'listdir("{path}")'
elif is_foreach_execute(m_type):
m_spec['cmd'] = next((v for k, v in m_members if k == "cmd"), None)
m_spec['type_name'] = FOREACH_EXECUTE_TYPE
provider = next((v for k, v in m_members if k == "provider"), None)
if provider:
m_spec['provider'] = cls.from_object(provider)
else:
m_spec['provider'] = Spec(
name=ANONYMOUS_SPEC_NAME,
type=None,
type_name=NONE_TYPE,
repr='NONE PROVIDER')
m_spec['repr'] = 'foreach_execute("{cmd}", provider={provider})'
elif is_container_execute(m_type):
m_spec['cmd'] = next((v for k, v in m_members if k == "cmd"), None)
m_spec['type_name'] = CONTAINER_EXECUTE_TYPE
provider = next((v for k, v in m_members if k == "provider"), None)
if provider:
m_spec['provider'] = cls.from_object(provider)
else:
m_spec['provider'] = Spec(
name=ANONYMOUS_SPEC_NAME,
type=None,
type_name=NONE_TYPE,
repr='NONE PROVIDER')
m_spec['repr'] = 'container_execute("{cmd}", provider={provider})'
elif is_first_of(m_type):
m_spec['type_name'] = FIRST_OF_TYPE
deps = next((v for k, v in m_members if k == "deps"), None)
m_spec['deps'] = [cls.from_object(d) for d in deps]
deps_repr = ', '.join(['{0}'.format(d) for d in m_spec['deps']])
m_spec['repr'] = 'first_of([{0}])'.format(deps_repr)
elif is_command_with_args(m_type):
m_spec['type_name'] = COMMAND_WITH_ARGS_TYPE
m_spec['cmd'] = next((v for k, v in m_members if k == "cmd"), None)
provider = next((v for k, v in m_members if k == "provider"), None)
if provider:
m_spec['provider'] = cls.from_object(provider)
m_spec['repr'] = 'command_with_args("{cmd}", provider={provider})'
elif is_foreach_collect(m_type):
m_spec['type_name'] = FOREACH_COLLECT_TYPE
m_spec['path'] = next((v for k, v in m_members if k == 'path'), None)
provider = next((v for k, v in m_members if k == "provider"), None)
if provider:
m_spec['provider'] = cls.from_object(provider)
m_spec['repr'] = 'foreach_collect("{path}", provider={provider})'
elif is_container_collect(m_type):
m_spec['type_name'] = CONTAINER_COLLECT_TYPE
# container_collect is based on foreach_execute, it used obj.cmd as the path
m_spec['path'] = next((v for k, v in m_members if k == 'cmd'), None)
provider = next((v for k, v in m_members if k == "provider"), None)
if provider:
m_spec['provider'] = cls.from_object(provider)
m_spec['repr'] = 'container_collect("{path}", provider={provider})'
elif is_head(m_type):
m_spec['type_name'] = HEAD_TYPE
dep = next((v for k, v in m_members if k == "dep"), None)
m_spec['dep'] = cls.from_object(dep)
m_spec['repr'] = 'head({0})'.format(m_spec['dep'])
elif m_type is None:
m_spec['type_name'] = NONE_TYPE
m_spec['repr'] = 'NONE TYPE'
elif inspect.isfunction(m_type):
m_spec['type_name'] = FUNCTION_TYPE
f_members = inspect.getmembers(m_type)
m_spec['fxn_name'] = next((v for k, v in f_members if k == '__name__'), 'function')
m_spec['source'] = inspect.getsource(m_type)
m_spec['repr'] = '{fxn_name}()'
elif isinstance(m_type, list):
m_spec['type_name'] = LIST_TYPE
m_spec['list'] = m_type
m_spec['repr'] = '{list}'
elif isinstance(m_type, str):
m_spec['type_name'] = STRING_TYPE
m_spec['string'] = m_type
m_spec['repr'] = '{string}'
else:
raise AutologyError('Unsupported name {} object {}, please add it'.format(m_name, m_type))
m_spec['repr'] = ''.join(['{name} = ', m_spec['repr']]) if m_name != ANONYMOUS_SPEC_NAME else m_spec['repr']
return cls(**m_spec)
@property
def is_simple_command(self):
""" bool: True if this spec is a simple_command """
return self.get('type_name', UNKNOWN_TYPE) == SIMPLE_COMMAND_TYPE
@property
def is_simple_file(self):
""" bool: True if this spec is a simple_file """
return self.get('type_name', UNKNOWN_TYPE) == SIMPLE_FILE_TYPE
@property
def is_glob_file(self):
""" bool: True if this spec is a glob_file """
return self.get('type_name', UNKNOWN_TYPE) == GLOB_FILE_TYPE
@property
def is_foreach_execute(self):
""" bool: True if this spec is a foreach_execute """
return self.get('type_name', UNKNOWN_TYPE) == FOREACH_EXECUTE_TYPE
@property
def is_first_file(self):
""" bool: True if this spec is a first_file """
return self.get('type_name', UNKNOWN_TYPE) == FIRST_FILE_TYPE
@property
def is_first_of(self):
""" bool: True if this spec is a first_of """
return self.get('type_name', UNKNOWN_TYPE) == FIRST_OF_TYPE
@property
def is_foreach_collect(self):
""" bool: True if this spec is a foreach_collect """
return self.get('type_name', UNKNOWN_TYPE) == FOREACH_COLLECT_TYPE
@property
def is_listdir(self):
""" bool: True if this spec is a listdir """
return self.get('type_name', UNKNOWN_TYPE) == LISTDIR_TYPE
@property
def is_command_with_args(self):
""" bool: True if this spec is a command_with_args """
return self.get('type_name', UNKNOWN_TYPE) == COMMAND_WITH_ARGS_TYPE
@property
def is_head(self):
""" bool: True if this spec is a head """
return self.get('type_name', UNKNOWN_TYPE) == HEAD_TYPE
@property
def is_function(self):
""" bool: True if this spec is a function """
return self.get('type_name', UNKNOWN_TYPE) == FUNCTION_TYPE
@property
def is_unknown_type(self):
""" bool: True if this spec has an unknown type """
return self.get('type_name', UNKNOWN_TYPE) == UNKNOWN_TYPE
[docs]
class RegistrySpecs(dict):
"""
Class to provide introspection for Registry objects in :py:class:`insights.specs.Specs`
Dictionary of `Spec` objects with spec names as the keys and a Spec object as each value.
Each value has the set of attributes::
name - Name of the spec
type - The spec object
filterable - Whether or not the Registry object is ``filterable``
multi_output - Whether or not the Registry object is ``multi_output``
raw - Whether or not the Registry object is ``raw``
"""
REGISTRY_REPR = '{name} = RegistryPoint(filterable={filterable}, multi_output={multi_output}, raw={raw})'
""" str: repr string for all Registry objects """
def __init__(self):
members = inspect.getmembers(insights.specs.Specs)
for m_name, m_type in members:
if m_name.startswith('__') or m_name == 'context_handlers':
# Don't care about dunder members or context handlers
continue
m_members = inspect.getmembers(m_type)
self.update({
m_name: Spec(
name=m_name,
type=m_type,
filterable=next((v for k, v in m_members if k == 'filterable'), None),
multi_output=next((v for k, v in m_members if k == 'multi_output'), None),
raw=next((v for k, v in m_members if k == 'raw'), None),
repr=self.REGISTRY_REPR
)
})
[docs]
def is_registered(self, spec_name):
""" bool: Whether ``spec_name`` is in Registry specs. """
return spec_name in self
[docs]
class DefaultSpecs(dict):
"""
Class to provide introspection for datasource objects in :py:class:`insights.specs.default.DefaultSpecs`
Dictionary of `Spec` objects with spec names as the keys and a Spec object as each value.
Each Spec has different attributes depending on the type of spec. See the
:py:meth:`Spec.from_object` factory method for more information.
"""
_SPECS_OBJECT_CLASS = InDefaultSpecs
""" obj: Datasource class to use for introspection """
def __init__(self):
for m_name, m_type in inspect.getmembers(self._SPECS_OBJECT_CLASS):
if (m_name.startswith('__') or m_name == 'context_handlers' or
m_name == 'registry' or str(m_type).startswith('insights.specs.Spec')):
# Don't care about dunder members or context handlers
# or registry entries
continue
m_spec = Spec.from_object(m_type, m_name)
self.update({m_name: m_spec})
[docs]
class InsightsArchiveSpecs(DefaultSpecs):
"""
Class to provide introspection for datasource objects in
:py:class:`insights.specs.insights_archive.InsightsArchiveSpecs`
Dictionary of `Spec` objects with spec names as the keys and a Spec object as each value.
Each Spec has different attributes depending on the type of spec. See the
:py:meth:`Spec.from_object` factory method for more information.
"""
_SPECS_OBJECT_CLASS = InArchiveSpecs
""" obj: Datasource class to use for introspection """
pass
[docs]
class SosSpecs(DefaultSpecs):
"""
Class to provide introspection for datasource objects in :py:class:`insights.specs.sos_archive.SosSpecs`
Dictionary of `Spec` objects with spec names as the keys and a Spec object as each value.
Each Spec has different attributes depending on the type of spec. See the
:py:meth:`Spec.from_object` factory method for more information.
"""
_SPECS_OBJECT_CLASS = InSosSpecs
""" obj: Datasource class to use for introspection """
pass
if __name__ == "__main__":
specs = DefaultSpecs()
for k, v in specs.items():
try:
print(v)
except Exception:
print('======= Error with spec: ', k)
print('repr_str :', v.repr_str)
for dk, dv in v.items():
print('\t', dk, ":", dv)
raise