Built-In Concurrency and Composing Custom Tasks

import inspect
from nornir.plugins.runners import ThreadedRunner

print(inspect.signature(ThreadedRunner.__init__))

Verified directly against the installed package.

(self, num_workers: int = 20)

Nornir's default runner (plugin="threaded") is a bounded thread pool of 20 workers -- the same bounded-concurrency idea Phase 3.3's ThreadPoolExecutor(max_workers=...) lesson already covered by hand.

import time
from nornir import InitNornir
from nornir.core.task import Task, Result

def slow_task(task: Task) -> Result:
  time.sleep(0.3)
  return Result(host=task.host, result="done")

for workers in (1, 20):
  nr = InitNornir(
      runner={"plugin": "threaded", "options": {"num_workers": workers}},
      inventory={"plugin": "SimpleInventory", "options": {
          "host_file": "inventory/hosts.yaml",
          "group_file": "inventory/groups.yaml",
          "defaults_file": "inventory/defaults.yaml",
      }},
  )
  start = time.time()
  nr.run(task=slow_task)
  elapsed = time.time() - start
  print(f"num_workers={workers}: {elapsed:.2f}s")

5 hosts, each task simulating 0.3s of work -- num_workers=1 forces true serial execution; num_workers=20 is Nornir's real default.

num_workers=1: 1.50s
num_workers=20: 0.30s

Real, measured output -- not invented. 1.50s = 5 x 0.3s, exactly serial. 0.30s = all 5 hosts run at once, exactly concurrent. A 4.97x speedup, measured directly against this module's own 5-device fleet.

def get_vlans(task: Task) -> Result:
  hostname = task.host.name
  try:
      vlans = simulate_connection(hostname)
  except (ConnectionError, PermissionError) as e:
      return Result(host=task.host, result=str(e), failed=True)
  return Result(host=task.host, result=vlans, failed=False)

simulate_connection() is the same Netmiko-shaped simulation pattern established in Your First Network Automation -- not a new mock framework, just a function returning realistic success/failure outcomes without a real device.

def run(self, task: Callable[..., Result], **kwargs: Any) -> MultiResult:
  ...
  r = self._start_children_task(task, **kwargs)
  if r.failed and self.host.close_connections_on_error:
      raise NornirSubTaskError(...)
  return self.results

Verified directly from the installed package's source. If the sub-task fails, task.run() raises NornirSubTaskError -- it does not return a normal Result you check with .failed afterward.

from nornir.core.exceptions import NornirSubTaskError

def validate_vlans(task: Task) -> Result:
  try:
      sub_result = task.run(task=get_vlans)
  except NornirSubTaskError as e:
      return Result(host=task.host, result=e.result.result, failed=True)

  required = REQUIRED_VLANS[task.host.name]
  actual = sub_result[0].result
  missing = [v for v in required if v not in actual]
  if missing:
      return Result(host=task.host, result=f"missing VLANs: {[v['id'] for v in missing]}", failed=False)
  return Result(host=task.host, result="already compliant", failed=False)

try/except NornirSubTaskError is required, not optional. e.result.result recovers the original failure message from get_vlans -- SW4/SW5's real failures, still readable after the exception.

SW1: failed=False result='already compliant'
SW2: failed=False result='missing VLANs: [10]'
SW3: failed=False result='missing VLANs: [30]'
SW4: failed=True result='SW4 did not respond'
SW5: failed=True result='SW5 rejected the credentials'

Real, verified output. SW4 and SW5's original failure messages (from get_vlans, raised via simulate_connection) survive intact through the caught NornirSubTaskError.