Nornir for Network Engineers
Capstone: A Nornir-Based Network Health Check
What you'll learn
- Combining inventory, filtering, task execution, and concurrency into one script
- Where a real device (Option A, nornir_netmiko) fits versus the local, no-device path (Option B)
- Reading a real, verified end-to-end run and its JSON report
- What's next: NAPALM, for multi-vendor abstraction beyond Cisco IOS
Listen to a guided explanation of this lesson.
Spoken narration isn't supported in this browser.
What this capstone combines
Every lesson in this module built one piece. This capstone runs all of them together, against the exact same VLAN desired-state data this curriculum has used since Phase 2.9.
- InitNornir()Lesson 3 — load hosts.yaml/groups.yaml/defaults.yaml
- Inject credentialsLesson 2 — from environment variables, never on disk
- nr.filter(site="HQ")Lesson 4 — demonstrated, not required for the full run
- nr.run(task=check_vlans)Lesson 3 + 6 — concurrent by default
- Option A: nornir_netmikoLesson 5 — real device
- Option B: simulate_connection()no device — this capstone's default path
- AggregatedResult -> JSON reportLesson 3 — structured results
Two options, same shape as every prior capstone
Option A: swap check_vlans's body for nr.run(task=netmiko_send_command, command_string="show vlan brief") against real devices, exactly as Lesson 5 verified. Option B (used below): simulate_connection(), the same Netmiko-shaped simulation pattern from Your First Network Automation -- no real device, but the exact same Nornir workflow around it.
The full script
import json
import os
import time
from nornir import InitNornir
from nornir.core.task import Task, Result
REQUIRED_VLANS = {
"SW1": [{"id": 10, "name": "USERS"}, {"id": 20, "name": "VOICE"}],
"SW2": [{"id": 10, "name": "USERS"}],
"SW3": [{"id": 10, "name": "USERS"}, {"id": 30, "name": "MGMT"}],
"SW4": [{"id": 10, "name": "USERS"}],
"SW5": [{"id": 10, "name": "USERS"}],
}
_SCENARIOS = {
"SW1": {"vlans": [{"id": 10, "name": "USERS"}, {"id": 20, "name": "VOICE"}], "outcome": "ok"},
"SW2": {"vlans": [], "outcome": "ok"},
"SW3": {"vlans": [{"id": 10, "name": "USERS"}], "outcome": "ok"},
"SW4": {"vlans": [], "outcome": "unreachable"},
"SW5": {"vlans": [], "outcome": "auth-failed"},
}
def simulate_connection(hostname):
# Option B: the same Netmiko-shaped simulation pattern from
# Your First Network Automation -- no real device, same workflow.
time.sleep(0.3)
scenario = _SCENARIOS[hostname]
if scenario["outcome"] == "unreachable":
raise ConnectionError(f"{hostname} did not respond")
if scenario["outcome"] == "auth-failed":
raise PermissionError(f"{hostname} rejected the credentials")
return scenario["vlans"]
def check_vlans(task: Task) -> Result:
hostname = task.host.name
try:
actual_vlans = simulate_connection(hostname)
except (ConnectionError, PermissionError) as e:
return Result(host=task.host, result=str(e), failed=True)
required = REQUIRED_VLANS[hostname]
missing = [v for v in required if v not in actual_vlans]
if missing:
return Result(
host=task.host,
result=f"missing VLANs: {[v['id'] for v in missing]}",
failed=False,
changed=True,
)
return Result(host=task.host, result="already compliant", failed=False, changed=False)
nr = InitNornir(
inventory={
"plugin": "SimpleInventory",
"options": {
"host_file": "inventory/hosts.yaml",
"group_file": "inventory/groups.yaml",
"defaults_file": "inventory/defaults.yaml",
},
},
)
nr.inventory.defaults.username = os.environ.get("NORNIR_USERNAME", "")
nr.inventory.defaults.password = os.environ.get("NORNIR_PASSWORD", "")
hq_only = nr.filter(site="HQ")
print(f"HQ-only run would target: {list(hq_only.inventory.hosts.keys())}")
agg = nr.run(task=check_vlans)
report = {"compliant": [], "non_compliant": [], "unreachable": [], "auth_failed": []}
for hostname, multi_result in agg.items():
r = multi_result[0]
if not r.failed and not r.changed:
report["compliant"].append({"host": hostname, "detail": r.result})
elif not r.failed and r.changed:
report["non_compliant"].append({"host": hostname, "detail": r.result})
elif "did not respond" in r.result:
report["unreachable"].append({"host": hostname, "detail": r.result})
else:
report["auth_failed"].append({"host": hostname, "detail": r.result})
print("--- Summary ---")
print(f"Checked: {len(agg)} devices")
print(f"compliant: {len(report['compliant'])}")
print(f"non-compliant: {len(report['non_compliant'])}")
print(f"unreachable: {len(report['unreachable'])}")
print(f"auth-failed: {len(report['auth_failed'])}")
with open("nornir_report.json", "w") as f:
json.dump(report, f, indent=2)
print("Report written to nornir_report.json")The complete, real, verified capstone script. Reuses the exact REQUIRED_VLANS desired-state data from Phase 2.9 onward, and the simulate_connection() philosophy from Your First Network Automation.
Real, verified end-to-end output
HQ-only run would target: ['SW1', 'SW2', 'SW3']
--- Summary ---
Checked: 5 devices
compliant: 1
non-compliant: 2
unreachable: 1
auth-failed: 1
Report written to nornir_report.jsonReal, verified output from an actual run of the script above. The HQ-only filter line demonstrates Lesson 4's capability without narrowing the actual run -- the full check still covers all 5 devices.
{
"compliant": [
{ "host": "SW1", "detail": "already compliant" }
],
"non_compliant": [
{ "host": "SW2", "detail": "missing VLANs: [10]" },
{ "host": "SW3", "detail": "missing VLANs: [30]" }
],
"unreachable": [
{ "host": "SW4", "detail": "SW4 did not respond" }
],
"auth_failed": [
{ "host": "SW5", "detail": "SW5 rejected the credentials" }
]
}Real, verified file content -- grep-confirmed to contain no credential of any kind, matching the runtime-injection pattern from Lesson 2.
What's next
One vendor's CLI so far
Every device in this curriculum has spoken Cisco IOS. NAPALM, the next planned module, addresses a different problem: a single, consistent API across multiple vendors (Cisco, Juniper, Arista, and others) instead of vendor-specific commands. Nornir itself doesn't change -- nornir_napalm is simply a different task plugin, the same role nornir_netmiko has played throughout this module.
Key takeaways
- This capstone combines every piece of the module: InitNornir(), runtime credential injection, nr.filter(), nr.run() with concurrency, and AggregatedResult-based reporting -- all verified together in one real, runnable script.
- Option A (nornir_netmiko against a real device) and Option B (simulate_connection(), used here) differ only in check_vlans's body -- the surrounding Nornir workflow is identical.
- Real, verified run: 1 compliant, 2 non-compliant, 1 unreachable, 1 auth-failed across the 5-device fleet -- written to nornir_report.json with zero credentials on disk.
- Next: NAPALM, for a consistent API across multiple vendors -- a different task plugin, the same Nornir workflow.