import augeas
from jadi import interface
[docs]class AugeasError(Exception):
def __init__(self, aug):
self.message = None
self.data = {}
aug.dump('/')
for ep in aug.match('/augeas//error'):
self.message = aug.get(ep + '/message')
for p in aug.match(ep + '/*'):
self.data[p.split('/')[-1]] = aug.get(p)
def __str__(self):
return self.message
[docs]class Augeas(augeas.Augeas):
"""
A smarter and faster wrapper around :class:`augeas.Augeas`.augeas
For faster startup, no modules and lenses are preloaded::
aug = Augeas(modules=[{
'name': 'Interfaces', # module name
'lens': 'Interfaces.lns', # lens name
'incl': [ # included files list
self.path,
self.path + '.d/*',
]
}])
Don't forget to call :func:`.load()` afterwards.
"""
def __init__(self, modules=[], loadpath=None):
augeas.Augeas.__init__(self, loadpath=loadpath, flags=augeas.Augeas.NO_MODL_AUTOLOAD | augeas.Augeas.NO_LOAD)
for module in modules:
path = f'/augeas/load/{module["name"]}'
self.set(f'{path}/lens', module['lens'])
for index, incl in enumerate(module['incl']):
self.set(f'{path}/incl[{index+1}]', incl)
def __enc(self, v):
return v
[docs] def match(self, path):
return augeas.Augeas.match(self, self.__enc(path))
[docs] def get(self, path):
return augeas.Augeas.get(self, self.__enc(path))
[docs] def set(self, path, value):
augeas.Augeas.set(self, self.__enc(path), self.__enc(value))
[docs] def setd(self, path, value, default=None):
"""
Sets `path` to `value`, or removes `path` if `value == default`
"""
if value is not None:
self.set(path, value)
if value == default:
self.remove(path)
[docs] def raise_error(self):
"""
Extracts error information from Augeas tree and raises :exc:`AugeasError`
"""
raise AugeasError(self)
[docs] def save(self):
try:
augeas.Augeas.save(self)
except IOError:
self.raise_error()
[docs] def dump(self, path):
"""
Dumps contents under `path` to stdout.
"""
for sp in self.match(path + '/*'):
print(sp, '=', self.get(sp))
[docs]@interface
class AugeasEndpoint():
"""
Implement this to provide Augeas trees to the frontend.
"""
id = None
def __init__(self, context):
self.context = context
[docs] def get_augeas(self):
"""
Should return a ready-to-use :class:`Augeas`
"""
raise NotImplementedError
[docs] def get_root_path(self):
"""
Should return an Augeas path of the root node to be provided to the frontend.
"""
raise NotImplementedError