How to load the inventory from an async application

Use AInitNornir when your program already runs an event loop, for example a web service, a bot or an async test suite. nr = await AInitNornir(...) takes the same arguments as InitNornir and returns the same Nornir object, but the event loop keeps serving other work while the inventory loads.

Build Nornir with an existing inventory plugin

Existing inventory plugins work with AInitNornir, SimpleInventory included. A synchronous plugin is constructed and loaded in a worker thread, so a slow inventory source does not stop the event loop. A plugin that must run in the main thread, for example to install a signal handler, fails with AInitNornir: keep using InitNornir for it, or add an asynchronous aload to it. Paths in this example are relative to the notebook working directory, docs/howto/.

import asyncio
import threading
from typing import Any

from nornir import AInitNornir, InitNornir
from nornir.core.exceptions import (
    InventoryPluginNotSyncError,
    TransformFunctionNotSyncError,
)
from nornir.core.inventory import Defaults, Groups, Host, Hosts, Inventory
from nornir.core.plugins.inventory import (
    AsyncTransformFunctionRegister,
    CapabilityInventoryPluginRegister,
    InventoryCapability,
)

nr = await AInitNornir(config_file="async_inventory/config.yaml")
for host in nr.inventory.hosts.values():
    print(f"{host.name}: {host.hostname} ({host.platform}, {host['site']})")

edge-1: 192.0.2.1 (eos, lab-a)
edge-2: 192.0.2.2 (eos, lab-b)

Load from an asynchronous source

An inventory plugin whose API client is asynchronous implements the capability-aware contract: it declares asyncio support in get_capabilities and implements aload. AInitNornir awaits aload on the running loop. The class below stands in for a plugin that queries an inventory API; a real plugin is registered through the nornir.plugins.capability_inventory entry-point group instead of register.

API_DEVICES = [
    {"name": "core-1", "address": "198.51.100.1", "platform": "junos"},
    {"name": "core-2", "address": "198.51.100.2", "platform": "junos"},
]


class LabApiInventory:
    def __init__(self, site: str) -> None:
        self.site = site

    def get_capabilities(self) -> frozenset[InventoryCapability]:
        return frozenset({"asyncio"})

    async def aload(self) -> Inventory:
        await asyncio.sleep(0)  # an awaited HTTP or GraphQL call in a real plugin
        hosts = Hosts(
            {
                device["name"]: Host(
                    name=device["name"],
                    hostname=device["address"],
                    platform=device["platform"],
                    data={"site": self.site},
                )
                for device in API_DEVICES
            }
        )
        return Inventory(hosts=hosts, groups=Groups(), defaults=Defaults())


CapabilityInventoryPluginRegister.register("lab_api", LabApiInventory)

api_nr = await AInitNornir(inventory={"plugin": "lab_api", "options": {"site": "lab-c"}})
for host in api_nr.inventory.hosts.values():
    print(f"{host.name}: {host.hostname} ({host.platform}, {host['site']})")

core-1: 198.51.100.1 (junos, lab-c)
core-2: 198.51.100.2 (junos, lab-c)

Enrich hosts with an async transform function

A transform function defined with async def is awaited once per host, in inventory order, one host at a time. Use it when the data comes from a source with an asynchronous client, such as a secrets store. A synchronous transform function also works with AInitNornir; it runs in a worker thread.

SECRETS = {"core-1": "s3cr3t-1", "core-2": "s3cr3t-2"}


async def read_password(host: Host, **kwargs: Any) -> None:
    await asyncio.sleep(0)  # an awaited call to the secrets store
    host.password = SECRETS[host.name]
    print(f"password set for {host.name}")


AsyncTransformFunctionRegister.register("read_password", read_password)

enriched_nr = await AInitNornir(
    inventory={
        "plugin": "lab_api",
        "options": {"site": "lab-c"},
        "transform_function": "read_password",
    }
)
print(all(host.password for host in enriched_nr.inventory.hosts.values()))

password set for core-1
password set for core-2
True

Errors from InitNornir

InitNornir does not run asynchronous code. When the configured plugin only declares asyncio support, or the transform function is defined with async def, it raises a named error before anything is loaded. Both errors are subclasses of SyncAsyncMismatchError.

try:
    InitNornir(inventory={"plugin": "lab_api", "options": {"site": "lab-c"}})
except InventoryPluginNotSyncError as exc:
    print(f"{type(exc).__name__}: {exc}")

try:
    InitNornir(
        config_file="async_inventory/config.yaml",
        inventory={"transform_function": "read_password"},
    )
except TransformFunctionNotSyncError as exc:
    print(f"{type(exc).__name__}: {exc}")

InventoryPluginNotSyncError: Inventory plugin 'lab_api' only declares asyncio support and cannot be loaded by InitNornir(). Use 'nr = await AInitNornir(...)' instead, or ask the plugin author for synchronous support.
TransformFunctionNotSyncError: Transform function 'read_password' is defined with 'async def' and cannot be applied by InitNornir(). Use 'nr = await AInitNornir(...)' instead, or define a synchronous transform function.

Cancellation

If the application cancels AInitNornir, for example on a startup timeout, the cancellation reaches an asynchronous plugin’s aload immediately. A synchronous plugin running in the worker thread cannot be interrupted: AInitNornir returns control at once, but the thread keeps running until load returns, and its result is discarded. Synchronous plugins should therefore set timeouts on their network calls.