← Interview Prep

Nornir — Practice

Hands-on Nornir: InitNornir + inventory files, fleet-wide show commands, F-object filtering, config backup with grouped subtasks, Jinja2 render + NAPALM dry-run/commit, failure handling & retries, and pytest.

Theory into muscle memory: initialize Nornir, structure an inventory, then run the everyday flows — collect, filter, template, and deploy with a dry-run. Companion to Nornir — Theory; the same SoT→render→ diff→deploy shape appears in Python Automation Workflows.

Initialize & inventory

# config.yaml
inventory:
  plugin: SimpleInventory
  options:
    host_file: "inventory/hosts.yaml"
    group_file: "inventory/groups.yaml"
    defaults_file: "inventory/defaults.yaml"
runner:
  plugin: threaded
  options: { num_workers: 20 }

# inventory/hosts.yaml
r1:
  hostname: 10.0.0.1
  platform: ios          # netmiko/napalm driver name
  groups: [edge]
  data: { site: lhr, role: edge }
r2:
  hostname: 10.0.0.2
  platform: ios
  groups: [edge]

# inventory/groups.yaml
edge:
  username: admin
  password: "{{ env 'NET_PASS' }}"   # or inject at runtime, never commit secrets

from nornir import InitNornir
nr = InitNornir(config_file="config.yaml")

Collect: a show command fleet-wide

from nornir_netmiko import netmiko_send_command
from nornir_utils.plugins.functions import print_result

result = nr.run(task=netmiko_send_command, command_string="show version")
print_result(result)                      # per-host, structured
failed = list(nr.data.failed_hosts)       # who errored

Filter: target a subset

from nornir.core.filter import F

edge_lhr = nr.filter(F(groups__contains="edge") & F(site="lhr"))
edge_lhr.run(task=netmiko_send_command, command_string="show ip int brief")

Config backup (grouped subtasks)

from nornir.core.task import Task, Result

def backup(task: Task) -> Result:
    out = task.run(task=netmiko_send_command, command_string="show run")
    path = f"backups/{task.host.name}.cfg"
    with open(path, "w") as fh:
        fh.write(out.result)
    return Result(host=task.host, result=f"saved {path}")

nr.run(task=backup)

Render + deploy with a dry-run (NAPALM)

The safe pattern: render from a template, push in merge mode, review the diff, then commit.

from nornir_jinja2.plugins.tasks import template_file
from nornir_napalm.plugins.tasks import napalm_configure

def push_ntp(task: Task, dry_run: bool = True) -> Result:
    cfg = task.run(task=template_file,
                   template="ntp.j2", path="templates/").result
    task.run(task=napalm_configure, configuration=cfg, dry_run=dry_run)

nr.run(task=push_ntp, dry_run=True)    # shows the diff, changes nothing
# ... review ...
nr.run(task=push_ntp, dry_run=False)   # commit

# templates/ntp.j2
{% for s in host['ntp_servers'] %}
ntp server {{ s }}
{% endfor %}

Failure handling & retries

result = nr.run(task=push_ntp, dry_run=False)
if result.failed:
    for host, mr in result.items():
        if mr.failed:
            print(host, mr.exception)
# only the failed hosts, to retry:
retry = nr.filter(filter_func=lambda h: h.name in result.failed_hosts)

• A failed host is skipped by later tasks in the same chain — re-filter to retry just those.
• It's plain Python, so wrap in try/except, add backoff, or gate on result.changed.

Test it (so CI can gate)

# tests/test_render.py — pure functions are easy to unit-test
def render_ntp(servers):
    return "\n".join(f"ntp server {s}" for s in servers)

def test_render_ntp():
    assert render_ntp(["10.0.0.53"]) == "ntp server 10.0.0.53"

Nornir tasks are ordinary functions, so pytest covers your logic and CI can block a bad change — the same gate you'd wire for Ansible in CI/CD.