import re
from lxml import etree
from lxml.builder import E
[docs]class _RpcMetaExec(object):
# -----------------------------------------------------------------------
# CONSTRUCTOR
# -----------------------------------------------------------------------
[docs] def __init__(self, junos):
"""
~PRIVATE CLASS~
creates an RPC meta-executor object bound to the provided
ez-netconf :junos: object
"""
self._junos = junos
# -----------------------------------------------------------------------
# get_config
# -----------------------------------------------------------------------
[docs] def get_config(self, filter_xml=None, options={}):
"""
retrieve configuration from the Junos device
:filter_xml: is options, defines what to retrieve. if omitted then the entire configuration is returned
:options: is a dict, creates attributes for the RPC
"""
rpc = E('get-configuration', options)
if filter_xml is not None:
# wrap the provided filter with toplevel <configuration> if
# it does not already have one
cfg_tag = 'configuration'
at_here = rpc if cfg_tag == filter_xml.tag else E(cfg_tag)
at_here.append(filter_xml)
if at_here is not rpc: rpc.append(at_here)
return self._junos.execute(rpc)
# -----------------------------------------------------------------------
# load_config
# -----------------------------------------------------------------------
[docs] def load_config(self, contents, **options):
"""
loads :contents: onto the Junos device, does not commit the change.
:options: is a dictionary of XML attributes to set within the <load-configuration> RPC.
The :contents: are interpreted by the :options: as follows:
format='text' and action='set', then :contents: is a string containing a series of "set" commands
format='text', then :contents: is a string containing Junos configuration in curly-brace/text format
<otherwise> :contents: is XML structure
"""
rpc = E('load-configuration', options)
if ('action' in options) and (options['action'] == 'set'):
rpc.append(E('configuration-set', contents))
elif ('format' in options) and (options['format'] == 'text'):
rpc.append(E('configuration-text', contents))
else:
# otherwise, it's just XML Element
if contents.tag != 'configuration':
etree.SubElement(rpc, 'configuration').append(contents)
else:
rpc.append(contents)
return self._junos.execute(rpc)
# -----------------------------------------------------------------------
# cli
# -----------------------------------------------------------------------
[docs] def cli(self, command, format='text'):
rpc = E('command', command)
if 'text' == format:
rpc.attrib['format'] = 'text'
return self._junos.execute(rpc)
# -----------------------------------------------------------------------
# method missing
# -----------------------------------------------------------------------
def __getattr__(self, rpc_cmd_name):
"""
metaprograms a function to execute the :rpc_cmd_name:
the caller will be passing (*vargs, **kvargs) on
execution of the meta function; these are the specific
rpc command arguments(**kvargs) and options bound
as XML attributes (*vargs)
"""
rpc_cmd = re.sub('_', '-', rpc_cmd_name)
def _exec_rpc(*vargs, **kvargs):
# create the rpc as XML command
rpc = etree.Element(rpc_cmd)
# kvargs are the command parameter/values
if kvargs:
for arg_name, arg_value in kvargs.items():
if arg_name not in ['dev_timeout', 'normalize']:
arg_name = re.sub('_', '-', arg_name)
if isinstance(arg_value, (tuple, list)):
for a in arg_value:
arg = etree.SubElement(rpc, arg_name)
if a is not True:
arg.text = a
else:
arg = etree.SubElement(rpc, arg_name)
if arg_value is not True:
arg.text = arg_value
# vargs[0] is a dict, command options like format='text'
if vargs:
for k, v in vargs[0].items():
if v is not True:
rpc.attrib[k] = v
# gather any decorator keywords
timeout = kvargs.get('dev_timeout')
normalize = kvargs.get('normalize')
dec_args = {}
if timeout is not None:
dec_args['dev_timeout'] = timeout
if normalize is not None:
dec_args['normalize'] = normalize
# now invoke the command against the
# associated :junos: device and return
# the results per :junos:execute()
return self._junos.execute(rpc, **dec_args)
# metabind help() and the function name to the :rpc_cmd_name:
# provided by the caller ... that's about all we can do, yo!
_exec_rpc.__doc__ = rpc_cmd
_exec_rpc.__name__ = rpc_cmd_name
# return the metafunction that the caller will in-turn invoke
return _exec_rpc
# -----------------------------------------------------------------------
# callable
# -----------------------------------------------------------------------
def __call__(self, rpc_cmd, **kvargs):
"""
callable will execute the provided :rpc_cmd: against the
attached :junos: object and return the RPC response per
:junos:execute()
kvargs is simply passed 'as-is' to :junos:execute()
"""
return self._junos.execute(rpc_cmd, **kvargs)