Python for Network Engineers — Part 11: Nornir — Parallel Automation at Scale
Every script in this series has had the same shape: load inventory, loop over devices, do something, collect results. The loop is sequential by default — device 2 doesn’t start until device 1 finishes. With three lab devices that’s fine. With 300 production devices, a script that takes 2 seconds per device runs for 10 minutes.
Nornir solves the concurrency problem without requiring you to write threading or async code. You write a task function — the thing to do on one device — and Nornir runs it across your entire inventory in parallel. Results come back in a structured object you can inspect, filter, and report on.
Nornir is also the integration layer. The Netmiko plugin wraps Part 5. The NAPALM plugin wraps Part 10. The Jinja2 plugin renders the templates from Part 8. You write the business logic once, and Nornir handles concurrency, inventory management, result collection, and error aggregation.
Installing Nornir
uv pip install nornir nornir-utils nornir-netmiko nornir-napalmnornir— the core frameworknornir-utils— helper tasks (print results, etc.)nornir-netmiko— Netmiko integrationnornir-napalm— NAPALM integration
Inventory
Nornir has its own YAML inventory format, separate from the one we built in Part 7. It uses three files:
hosts.yaml — one entry per device:
spine-01:
hostname: 192.168.100.10
username: admin
password: netauto
platform: eos
groups:
- spines
data:
role: spine
loopback_ip: 10.255.0.1/32
leaf-01:
hostname: 192.168.100.11
username: admin
password: netauto
platform: eos
groups:
- leaves
data:
role: leaf
loopback_ip: 10.255.0.2/32
vlans: [10, 20]
leaf-02:
hostname: 192.168.100.12
username: admin
password: netauto
platform: eos
groups:
- leaves
data:
role: leaf
loopback_ip: 10.255.0.3/32
vlans: [10, 20]groups.yaml — shared settings for groups of devices:
spines:
platform: eos
data:
bgp_asn: 65000
leaves:
platform: eos
data:
bgp_asn: 65001defaults.yaml — defaults applied to all hosts:
username: admin
password: netauto
data:
ntp_server: 192.168.100.1Values are resolved in order: host → group → defaults. A host’s own value always wins; missing values fall through to the group, then to defaults. This mirrors how Ansible variables work, but in plain YAML without special syntax.
config.yaml — tells Nornir where to find the inventory files:
inventory:
plugin: SimpleInventory
options:
host_file: hosts.yaml
group_file: groups.yaml
defaults_file: defaults.yaml
runner:
plugin: threaded
options:
num_workers: 10num_workers controls parallelism — 10 means up to 10 devices simultaneously. For a fleet of hundreds, 50–100 is typical; higher values hit SSH connection limits on managed devices.
Initialising Nornir
from nornir import InitNornir
nr = InitNornir(config_file="config.yaml")
print(f"Loaded {len(nr.inventory.hosts)} hosts")Or inline (without a config file, useful for scripts):
from nornir import InitNornir
from nornir.core.inventory import Host, Group, Defaults, Inventory
nr = InitNornir(
runner={"plugin": "threaded", "options": {"num_workers": 10}},
inventory={
"plugin": "SimpleInventory",
"options": {
"host_file": "hosts.yaml",
"group_file": "groups.yaml",
"defaults_file": "defaults.yaml",
},
},
)Task Functions
A task function is a regular Python function that Nornir calls once per host. It receives a Task object — which gives you access to the current host’s data — and returns a Result object.
from nornir.core.task import Task, Result
def get_hostname(task: Task) -> Result:
"""Return the hostname from the inventory."""
return Result(
host=task.host,
result=task.host.name,
)Run it across all hosts:
from nornir_utils.plugins.functions import print_result
results = nr.run(task=get_hostname)
print_result(results)nr.run() returns an AggregatedResult — a dict-like object keyed by hostname, where each value is a MultiResult (a list of results from this host’s task chain).
print_result() from nornir_utils prints a formatted summary:
get_hostname*******************************************************************
* leaf-01 ** changed : False ***************************************************
vvvv get_hostname ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv INFO
leaf-01
^^^^ END get_hostname ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
* leaf-02 ** changed : False ***************************************************
...Netmiko Plugin
The Nornir Netmiko plugin provides netmiko_send_command and netmiko_send_config tasks:
from nornir_netmiko import netmiko_send_command
# Run 'show version' on all hosts in parallel
results = nr.run(
task=netmiko_send_command,
command_string="show version",
)
for host, multi_result in results.items():
result = multi_result[0] # first (only) result in the chain
if not result.failed:
print(f"{host}: {len(result.result)} bytes of output")
else:
print(f"{host}: FAILED — {result.exception}")netmiko_send_config pushes a list of config commands:
from nornir_netmiko import netmiko_send_config
results = nr.run(
task=netmiko_send_config,
config_commands=[
"ntp server 192.168.1.1",
"ntp server 192.168.1.2",
],
)Writing Custom Tasks with Netmiko
More commonly, you write a task function that calls Netmiko tasks internally using task.run():
from nornir.core.task import Task, Result
from nornir_netmiko import netmiko_send_command
def collect_version(task: Task) -> Result:
"""Collect and parse show version output."""
result = task.run(
task=netmiko_send_command,
command_string="show version",
use_textfsm=True, # NTC-templates integration
)
parsed = result.result
if isinstance(parsed, list) and parsed:
version = parsed[0].get("VERSION") or parsed[0].get("eos_version", "unknown")
else:
version = "parse_failed"
return Result(
host=task.host,
result={"version": version, "host": task.host.hostname},
)
results = nr.run(task=collect_version)
for host, multi_result in results.items():
if not multi_result.failed:
print(f"{host}: {multi_result.result}")NAPALM Plugin
The NAPALM plugin provides napalm_get and napalm_configure tasks:
from nornir_napalm.plugins.tasks import napalm_get
# Collect facts from all devices simultaneously
results = nr.run(task=napalm_get, getters=["facts", "interfaces"])
for host, multi_result in results.items():
if not multi_result.failed:
data = multi_result.result
facts = data.get("facts", {})
intfs = data.get("interfaces", {})
up = sum(1 for d in intfs.values() if d["is_up"])
print(f"{host}: {facts.get('os_version')} — {up}/{len(intfs)} interfaces up")napalm_configure pushes a config and commits:
from nornir_napalm.plugins.tasks import napalm_configure
config = """
ntp server 192.168.1.1
ntp server 192.168.1.2
"""
results = nr.run(
task=napalm_configure,
configuration=config,
replace=False, # merge, not replace
dry_run=True, # show diff only, don't commit
)
print_result(results)dry_run=True runs the full diff but doesn’t commit — useful for review workflows.
Filtering Inventory
Running against a subset of hosts is one of Nornir’s most useful features:
# Filter to spine nodes only
spines = nr.filter(F(groups__contains="spines"))
results = spines.run(task=napalm_get, getters=["facts"])
# Filter by host data
leaves = nr.filter(F(data__role="leaf"))
# Filter by hostname pattern
target = nr.filter(filter_func=lambda h: "spine" in h.name)F comes from nornir.core.filter:
from nornir.core.filter import FFilters can be combined:
# Leaf nodes in the production group with a specific VLAN
prod_leaves = nr.filter(
F(groups__contains="leaves") & F(groups__contains="production")
)Filtering is how you build operational playbooks: deploy to spine tier first, validate, then deploy to leaves. Nornir’s filter API makes that explicit and auditable.
Result Handling
The AggregatedResult from nr.run() gives you programmatic access to everything:
results = nr.run(task=collect_version)
# Check if anything failed
if results.failed:
print(f"FAILED hosts: {[h for h, r in results.items() if r.failed]}")
# Iterate over results
for host, multi_result in results.items():
if multi_result.failed:
print(f" {host}: FAILED — {multi_result.exception}")
else:
data = multi_result.result
print(f" {host}: {data}")
# Separate failed from succeeded
succeeded = {h: r for h, r in results.items() if not r.failed}
failed = {h: r for h, r in results.items() if r.failed}
print(f"OK: {len(succeeded)}, Failed: {len(failed)}")result.changed is True if the task made a change (Nornir tracks this for you when using NAPALM configure tasks). Use it to count changes and flag unchanged hosts:
changed = [h for h, r in results.items() if r.changed]
unchanged = [h for h, r in results.items() if not r.changed and not r.failed]
print(f"Changed: {changed}")Capstone: Full Nornir Automation Script
A complete script that:
- Collects version and interface state from all lab devices in parallel via NAPALM
- Reports a summary to stdout
- Saves JSON output for further processing
"""Full Nornir automation script — parallel collection via NAPALM."""
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from nornir import InitNornir
from nornir.core.task import Task, Result
from nornir.core.filter import F
from nornir_napalm.plugins.tasks import napalm_get
from nornir_utils.plugins.functions import print_result
OUTPUT_FILE = "nornir_audit.json"
def collect_state(task: Task) -> Result:
"""Collect facts and interface state via NAPALM."""
result = task.run(task=napalm_get, getters=["facts", "interfaces", "interfaces_ip"])
data = result.result
facts = data.get("facts", {})
intfs = data.get("interfaces", {})
ip_info = data.get("interfaces_ip", {})
up_interfaces = [n for n, d in intfs.items() if d["is_up"]]
ip_summary = {
intf: list(addrs.get("ipv4", {}).keys())
for intf, addrs in ip_info.items()
if addrs.get("ipv4")
}
return Result(
host=task.host,
result={
"hostname": facts.get("hostname", task.host.name),
"vendor": facts.get("vendor"),
"model": facts.get("model"),
"os_version": facts.get("os_version"),
"uptime": facts.get("uptime"),
"interface_count": len(intfs),
"up_interfaces": up_interfaces,
"ip_addresses": ip_summary,
},
)
def main():
nr = InitNornir(config_file="config.yaml")
print(f"\nRunning against {len(nr.inventory.hosts)} host(s)...\n")
results = nr.run(task=collect_state)
# Summary table
print(f"\n{'Host':<12} {'Vendor':<10} {'Version':<14} {'Up/Total':<10} {'Status'}")
print("─" * 65)
output_devices = []
for host, multi_result in results.items():
if multi_result.failed:
print(f"{host:<12} {'—':<10} {'—':<14} {'—':<10} FAILED")
else:
d = multi_result.result
up = len(d["up_interfaces"])
total = d["interface_count"]
print(
f"{host:<12} "
f"{(d['vendor'] or '—'):<10} "
f"{(d['os_version'] or '—'):<14} "
f"{up}/{total:<8} OK"
)
output_devices.append({
"name": host,
"collected_at": datetime.now(timezone.utc).isoformat(),
**d,
})
# Save JSON
output = {
"run_at": datetime.now(timezone.utc).isoformat(),
"collected": len(output_devices),
"failed": [h for h, r in results.items() if r.failed],
"devices": output_devices,
}
Path(OUTPUT_FILE).write_text(json.dumps(output, indent=2), encoding="utf-8")
print(f"\nResults saved to {OUTPUT_FILE}")
if results.failed:
sys.exit(1)
if __name__ == "__main__":
main()Run it:
python3 nornir_audit.pyRunning against 3 host(s)...
Host Vendor Version Up/Total Status
─────────────────────────────────────────────────────────────────
spine-01 Arista 4.32.0F 2/5 OK
leaf-01 Arista 4.32.0F 3/6 OK
leaf-02 Arista 4.32.0F 3/6 OK
Results saved to nornir_audit.jsonAll three devices queried simultaneously — total runtime is the time for the slowest single device, not the sum.
Adding a Config Deploy Task
Nornir makes the full collect → render → deploy → verify cycle straightforward:
from jinja2 import Environment, FileSystemLoader
from nornir_napalm.plugins.tasks import napalm_configure
env = Environment(loader=FileSystemLoader("templates"), trim_blocks=True, lstrip_blocks=True)
def render_and_deploy(task: Task) -> Result:
"""Render a Jinja2 template and deploy via NAPALM."""
template = env.get_template(f"{task.host.platform}_config.j2")
rendered = template.render(device=task.host.data)
task.run(
task=napalm_configure,
configuration=rendered,
replace=True, # full replace
dry_run=False, # commit for real
)
return Result(host=task.host, changed=True, result="deployed")
# Deploy to spines first, then leaves
spines = nr.filter(F(groups__contains="spines"))
leaf_nr = nr.filter(F(groups__contains="leaves"))
print("Deploying to spines...")
results = spines.run(task=render_and_deploy)
if results.failed:
print("Spine deployment failed — aborting before leaves")
sys.exit(1)
print("Deploying to leaves...")
results = leaf_nr.run(task=render_and_deploy)
print_result(results)The sequential tier deployment (spines before leaves) is explicit and readable — a significant advantage over tools where execution order is implicit or requires special syntax.
When to Use Nornir
Nornir adds value when:
- You need to run the same task across more than a handful of devices
- Parallelism matters (more than ~10 devices)
- You want structured result aggregation and per-host error handling
- You’re building a production tool that others will use and need to maintain
For quick one-off scripts against a few devices, the direct Netmiko/NAPALM approach from earlier parts is less overhead. Nornir’s setup cost (inventory files, config file, task function structure) pays off at scale.
What’s Coming in Part 12
The final post in the series looks at how AI tools fit into the automation workflows we’ve built — not as a replacement for the Python skills you’ve developed, but as an accelerator. We cover AI-assisted config review, fleet-wide parsing for commands without templates, and building a minimal MCP server that lets an AI agent call your automation scripts directly.
This post uses the same Containerlab topology as Parts 5–10. No new lab infrastructure required.