Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def echo_task(task, msg="Nornir"):
return Result(
host=task.host,
result="Hello from {}".format(msg),
output="Hello from {}".format(msg),
)
def load_data(task):
data = {"os": "Linux", "services": ["http", "smtp", "dns"]}
return Result(host=task.host, result=data)
def get_version(task):
result = task.run(task=netmiko_send_command, command_string="show version")
return Result(host=task.host, result=result)
def my_task(task):
changed = False
if task.host.groups[0] == "ios":
cmd = "show run | i hostname"
task.run(task=networking.netmiko_send_command, command_string=cmd)
changed = True
elif task.host.groups[0] == "junos":
cmd = "show configuration system host-name "
task.run(task=networking.netmiko_send_command, command_string=cmd)
changed = True
return Result(host=task.host, changed=changed, failed=False, result="ran command")
* diff (``string``): change in the system
"""
device = task.host.get_connection("napalm")
if replace:
device.load_replace_candidate(filename=filename, config=configuration)
else:
device.load_merge_candidate(filename=filename, config=configuration)
diff = device.compare_config()
dry_run = task.is_dry_run(dry_run)
if not dry_run and diff:
device.commit_config()
else:
device.discard_config()
return Result(host=task.host, diff=diff, changed=len(diff) > 0)
* result (``dict``): dictionary with the result of the getter
"""
device = task.host.get_connection("napalm")
getters_options = getters_options or {}
if isinstance(getters, str):
getters = [getters]
result = {}
for g in getters:
options = copy.deepcopy(kwargs)
options.update(getters_options.get(g, {}))
getter = g if g.startswith("get_") else "get_{}".format(g)
method = getattr(device, getter)
result[g] = method(**options)
return Result(host=task.host, result=result)
Arguments:
host (:obj:`nornir.core.inventory.Host`): Host we are operating with. Populated right
before calling the ``task``
nornir(:obj:`nornir.core.Nornir`): Populated right before calling
the ``task``
Returns:
host (:obj:`nornir.core.task.MultiResult`): Results of the task and its subtasks
"""
self.host = host
self.nornir = nornir
try:
logger.debug("Host %r: running task %r", self.host.name, self.name)
r = self.task(self, **self.params)
if not isinstance(r, Result):
r = Result(host=host, result=r)
except NornirSubTaskError as e:
tb = traceback.format_exc()
logger.error(
"Host %r: task %r failed with traceback:\n%s",
self.host.name,
self.name,
tb,
)
r = Result(host, exception=e, result=str(e), failed=True)
except Exception as e:
tb = traceback.format_exc()
logger.error(
"Host %r: task %r failed with traceback:\n%s",
if commit_message == "":
commit_message = "File created with nornir"
pid = _get_repository(session, url, repository)
if action == "create":
diff = _create(
task, session, url, pid, filename, content, branch, commit_message, dry_run
)
elif action == "update":
diff = _update(
task, session, url, pid, filename, content, branch, commit_message, dry_run
)
elif action == "get":
diff = _get(task, session, url, pid, filename, destination, ref, dry_run)
return Result(host=task.host, diff=diff, changed=bool(diff))
Simple example with ``ordered_dict``::
> nr.run(task=load_json,
file="mydata.json")
file: path to the file containing the json file to load
Returns:
Result object with the following attributes set:
* result (``dict``): dictionary with the contents of the file
"""
with open(file, "r") as f:
data = json.loads(f.read())
return Result(host=task.host, result=data)
content: content you want to write
append: whether you want to replace the contents or append to it
Returns:
Result object with the following attributes set:
* changed (``bool``):
* diff (``str``): unified diff
"""
diff = _generate_diff(filename, content, append)
if not task.is_dry_run(dry_run):
mode = "a+" if append else "w+"
with open(filename, mode=mode) as f:
f.write(content)
return Result(host=task.host, diff=diff, changed=bool(diff))