Nautobot Nornir Sample Job

jobs.py

import logging

from datetime import datetime

from nautobot.extras.jobs import Job, MultiObjectVar, BooleanVar
from nautobot.extras.models import Tag
from nautobot.dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site, Platform, Region, Rack, RackGroup
from nautobot.tenancy.models import Tenant, TenantGroup

from nornir import InitNornir
from nornir.core.plugins.inventory import InventoryPluginRegister

from nautobot.dcim.filters import DeviceFilterSet

from nautobot_plugin_nornir.plugins.inventory.nautobot_orm import NautobotORMInventory
from nautobot_plugin_nornir.constants import NORNIR_SETTINGS
from nornir_netmiko.tasks import netmiko_send_command

InventoryPluginRegister.register("nautobot-inventory", NautobotORMInventory)


FIELDS = {
    "platform",
    "tenant_group",
    "tenant",
    "region",
    "site",
    "platform",
    "role",
    "rack",
    "rack_group",
    "manufacturer",
    "device_type",
}

def get_job_filter(data=None):
    """Helper function to return a the filterable list of OS's based on platform.slug and a specific custom value."""
    if not data:
        data = {}
    query = {}
    for field in FIELDS:
        if data.get(field):
            query[f"{field}_id"] = data[field].values_list("pk", flat=True)
    if data.get("device"):
        query.update({"id": data["device"].values_list("pk", flat=True)})
    return DeviceFilterSet(data=query, queryset=Device.objects.all()).qs


class SampleJob(Job):
    """Class definition."""

    tenant_group = MultiObjectVar(model=TenantGroup, required=False)
    tenant = MultiObjectVar(model=Tenant, required=False)
    region = MultiObjectVar(model=Region, required=False)
    site = MultiObjectVar(model=Site, required=False)
    rack_group = MultiObjectVar(model=RackGroup, required=False)
    rack = MultiObjectVar(model=Rack, required=False)
    role = MultiObjectVar(model=DeviceRole, required=False)
    manufacturer = MultiObjectVar(model=Manufacturer, required=False)
    platform = MultiObjectVar(model=Platform, required=False)
    device_type = MultiObjectVar(model=DeviceType, required=False, display_field="display_name")
    device = MultiObjectVar(model=Device, required=False)
    tag = MultiObjectVar(model=Tag, required=False)
    debug = BooleanVar(description="Enable for more verbose debug logging")

    class Meta:
        """Meta object boilerplate for job."""

        name = "Sample Job"
        description = "Run Sample Netmiko Job."

    def run(self, data, commit):  # pylint: disable=too-many-branches
        """Run config compliance report script."""

        nornir_obj = InitNornir(
            runner=NORNIR_SETTINGS.get("runner"),
            logging={"enabled": False},
            inventory={
                "plugin": "nautobot-inventory",
                "options": {
                    "credentials_class": NORNIR_SETTINGS.get("credentials"),
                    "params": NORNIR_SETTINGS.get("inventory_params"),
                    "queryset": get_job_filter(data),
                },
            },
        )

        result = nornir_obj.run(netmiko_send_command, command_string="show ip interface brief")

jobs = [SampleJob]