Netbox Get Interfaces Task

task api name: get_interfaces

Get Interfaces Sample Usage

NORFAB Netbox Get Interfaces Command Shell Reference

NorFab shell supports these command options for Netbox get_interfaces task:

nf#man tree netbox.get.interfaces
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── interfaces:    Query Netbox device interfaces data
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── *devices:    Devices to retrieve interface for
            ├── ip-addresses:    Retrieves interface IP addresses
            ├── inventory-items:    Retrieves interface inventory items
            └── dry-run:    Only return query content, do not run it
nf#

Python API Reference

Retrieve device interfaces from Netbox using GraphQL API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	str	Netbox instance name.	None
devices	list	List of devices to retrieve interfaces for.	None
ip_addresses	bool	If True, retrieves interface IPs. Defaults to False.	False
inventory_items	bool	If True, retrieves interface inventory items. Defaults to False.	False
dry_run	bool	If True, only return query content, do not run it. Defaults to False.	False

Returns:

Name	Type	Description
dict	Result	Dictionary keyed by device name with interface details.

Raises:

Type	Description
Exception	If no interfaces data is returned for the specified devices.

Source code in norfab\workers\netbox_worker.py

@Task()
def get_interfaces(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    ip_addresses: bool = False,
    inventory_items: bool = False,
    dry_run: bool = False,
    cache: Union[bool, str] = None,
) -> Result:
    """
    Retrieve device interfaces from Netbox using GraphQL API.

    Args:
        job: NorFab Job object containing relevant metadata
        instance (str, optional): Netbox instance name.
        devices (list, optional): List of devices to retrieve interfaces for.
        ip_addresses (bool, optional): If True, retrieves interface IPs. Defaults to False.
        inventory_items (bool, optional): If True, retrieves interface inventory items. Defaults to False.
        dry_run (bool, optional): If True, only return query content, do not run it. Defaults to False.

    Returns:
        dict: Dictionary keyed by device name with interface details.

    Raises:
        Exception: If no interfaces data is returned for the specified devices.
    """
    instance = instance or self.default_instance
    devices = devices or []
    ret = Result(
        task=f"{self.name}:get_interfaces",
        result={d: {} for d in devices},
        resources=[instance],
    )

    intf_fields = [
        "name",
        "enabled",
        "description",
        "mtu",
        "parent {name}",
        "mode",
        "untagged_vlan {vid name}",
        "vrf {name}",
        "tagged_vlans {vid name}",
        "tags {name}",
        "custom_fields",
        "last_updated",
        "bridge {name}",
        "child_interfaces {name}",
        "bridge_interfaces {name}",
        "member_interfaces {name}",
        "wwn",
        "duplex",
        "speed",
        "id",
        "device {name}",
    ]
    if self.nb_version[instance] >= (4, 2, 0):
        intf_fields.append("mac_addresses {mac_address}")
    else:
        intf_fields.append("mac_address")

    # add IP addresses to interfaces fields
    if ip_addresses:
        intf_fields.append(
            "ip_addresses {address status role dns_name description custom_fields last_updated tenant {name} tags {name}}"
        )

    # form interfaces query dictionary
    if self.nb_version[instance] >= (4, 3, 0):
        dlist = str(devices).replace("'", '"')  # swap quotes
        dfilt = "{device: {name: {in_list: " + dlist + "}}}"
    else:
        dfilt = {"device": devices}
    queries = {
        "interfaces": {
            "obj": "interface_list",
            "filters": dfilt,
            "fields": intf_fields,
        }
    }

    # add query to retrieve inventory items
    if inventory_items:
        if self.nb_version[instance] >= (4, 3, 0):
            dlist = str(devices).replace("'", '"')  # swap quotes
            inv_filters = (
                "{device: {name: {in_list: "
                + dlist
                + '}}, component_type: {app_label: {exact: "dcim"}}}'
            )
        else:
            inv_filters = {"device": devices, "component_type": "dcim.interface"}
        inv_fields = [
            "name",
            "component {... on InterfaceType {id}}",
            "role {name}",
            "manufacturer {name}",
            "custom_fields",
            "label",
            "description",
            "tags {name}",
            "asset_tag",
            "serial",
            "part_id",
        ]
        queries["inventor_items"] = {
            "obj": "inventory_item_list",
            "filters": inv_filters,
            "fields": inv_fields,
        }

    query_result = self.graphql(
        job=job, instance=instance, queries=queries, dry_run=dry_run
    )

    # return dry run result
    if dry_run:
        return query_result

    interfaces_data = query_result.result

    # exit if no Interfaces returned
    if interfaces_data is None or not interfaces_data.get("interfaces"):
        raise Exception(
            f"{self.name} - no interfaces data in '{interfaces_data}' returned by '{instance}' "
            f"for devices {', '.join(devices)}"
        )

    # process query results
    interfaces = interfaces_data.pop("interfaces")

    # process inventory items
    if inventory_items:
        inventory_items_list = interfaces_data.pop("inventor_items")
        # transform inventory items list to a dictionary keyed by intf_id
        inventory_items_dict = {}
        while inventory_items_list:
            inv_item = inventory_items_list.pop()
            # skip inventory items that does not assigned to components
            if inv_item.get("component") is None:
                continue
            intf_id = str(inv_item.pop("component").pop("id"))
            inventory_items_dict.setdefault(intf_id, [])
            inventory_items_dict[intf_id].append(inv_item)
        # iterate over interfaces and add inventory items
        for intf in interfaces:
            intf["inventory_items"] = inventory_items_dict.pop(intf["id"], [])

    # transform interfaces list to dictionary keyed by device and interfaces names
    while interfaces:
        intf = interfaces.pop()
        _ = intf.pop("id")
        device_name = intf.pop("device").pop("name")
        intf_name = intf.pop("name")
        if device_name in ret.result:  # Netbox issue #16299
            ret.result[device_name][intf_name] = intf

    return ret