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 Pynetbox REST 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
interface_list	list	List of interface names to retrieve.	None
interface_regex	str	Regex pattern to match interfaces by name.	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 REST filter params, do not run. Defaults to False.	False
cache	Union[None, bool, str]	True - use cache if up to date; False - skip cache; "refresh" - fetch and overwrite cache; "force" - use cache without staleness check	None

Returns:

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

This task performs no filtering of data returned by pynetbox.dcim.interfaces.filter, as a result use netbox REST API explorer to understand resulting data structure, REST API browser available at http://<netbox url>/api/dcim/interfaces/. In addition use nfcli to explore interfaces data:

nf#netbox get interfaces devices fceos4 interface-list eth201 | kv

netbox-worker-1.1.fceos4.eth201.display: eth201
netbox-worker-1.1.fceos4.eth201.device.name: fceos4
...

Can also pipe results through json, yaml, nested or pprint formatters to output results in certain format.

Source code in norfab\workers\netbox_worker\interfaces_tasks.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_interfaces(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    interface_list: Union[None, list] = None,
    interface_regex: Union[None, str] = None,
    ip_addresses: bool = False,
    inventory_items: bool = False,
    dry_run: bool = False,
    cache: Union[None, bool, str] = None,
    branch: str = None,
) -> Result:
    """
    Retrieve device interfaces from Netbox using Pynetbox REST 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.
        interface_list (list, optional): List of interface names to retrieve.
        interface_regex (str, optional): Regex pattern to match interfaces by name.
        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 REST filter params, do not run. Defaults to False.
        cache: ``True`` - use cache if up to date; ``False`` - skip cache;
            ``"refresh"`` - fetch and overwrite cache; ``"force"`` - use cache without staleness check

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

    This task performs no filtering of data returned by `pynetbox.dcim.interfaces.filter`,
    as a result use netbox REST API explorer to understand resulting data structure, REST
    API browser available at `http://<netbox url>/api/dcim/interfaces/`. In addition
    use nfcli to explore interfaces data:

    ```
    nf#netbox get interfaces devices fceos4 interface-list eth201 | kv

    netbox-worker-1.1.fceos4.eth201.display: eth201
    netbox-worker-1.1.fceos4.eth201.device.name: fceos4
    ...
    ```

    Can also pipe results through `json`, `yaml`, `nested` or `pprint`  formatters to output
    results in  certain format.
    """
    instance = instance or self.default_instance
    nb = self._get_pynetbox(instance, branch=branch)
    devices = devices or []
    cache = self.cache_use if cache is None else cache
    log.info(
        f"{self.name} - Get interfaces: Fetching interfaces for {len(devices)} device(s) from '{instance}'"
    )
    ret = Result(
        task=f"{self.name}:get_interfaces_pynetbox",
        result={d: {} for d in devices},
        resources=[instance],
    )
    filter_params = {}
    all_interfaces = []
    children_by_parent_id = {}  # parent_id -> [child intf, ...]
    member_intf_by_lag_id = {}  # lag_id    -> [member intf, ...]
    ip_by_intf_id = {}
    inv_by_intf_id = {}
    all_interfaces = []

    # build REST filter params
    if devices:
        filter_params["device"] = devices
    if interface_list:
        filter_params["name"] = interface_list
    if interface_regex:
        filter_params["name__regex"] = interface_regex

    if dry_run:
        ret.result = {"filter_params": filter_params}
        return ret

    job.event(f"retrieving interfaces for {len(devices)} device(s)")

    devices_to_fetch = list(devices)

    if cache == True or cache == "force":
        job.event(f"checking cache for {len(devices)} device(s)")
        # quick REST call to get current last_updated for all matching interfaces
        result = self.rest(
            job=job,
            instance=instance,
            api="dcim/interfaces",
            params={**filter_params, "fields": "name,last_updated,device"},
        )
        # build per-device last_updated map
        last_updated_by_device = {}
        for intf in result.result.get("results", []):
            dev_name = intf["device"]["name"]
            last_updated_by_device.setdefault(dev_name, {})[intf["name"]] = intf[
                "last_updated"
            ]

        self.cache.expire()  # remove expired items from cache
        devices_to_fetch = []
        for device_name, intf_last_updated in last_updated_by_device.items():
            device_cache_key = f"get_interfaces::{device_name}"
            if device_cache_key in self.cache and (
                cache == "force"
                or all(
                    self.cache[device_cache_key]
                    .get(intf_name, {})
                    .get("last_updated")
                    == lu
                    for intf_name, lu in intf_last_updated.items()
                )
            ):
                # serve requested interfaces from cache
                for intf_name in intf_last_updated.keys():
                    ret.result[device_name][intf_name] = self.cache[
                        device_cache_key
                    ][intf_name]
                job.event(
                    f"serving '{device_name}' interfaces from cache ({len(intf_last_updated)} interface(s))"
                )
            else:
                devices_to_fetch.append(device_name)
                job.event(
                    f"'{device_name}' cache miss or stale, fetching fresh data"
                )
    elif cache == False or cache == "refresh":
        pass  # fetch all devices fresh

    # build fetch filter params restricted to devices needing fresh data
    if devices and cache in (True, "force"):
        fetch_filter_params = {**filter_params, "device": devices_to_fetch}
    else:
        fetch_filter_params = filter_params

    # fetch all matching interfaces in one call
    if devices_to_fetch:
        job.event(
            f"fetching interfaces from NetBox for {len(devices_to_fetch)} device(s)"
        )
        all_interfaces = list(nb.dcim.interfaces.filter(**fetch_filter_params))
        job.event(f"retrieved {len(all_interfaces)} interface(s) from NetBox")

    if not all_interfaces and not any(ret.result.get(d) for d in devices):
        raise Exception(
            f"{self.name} - no interfaces data returned by '{instance}' "
            f"for devices {', '.join(devices)}"
        )

    # build relationship lookup maps from fetched data
    for intf in all_interfaces:
        if intf.parent:
            children_by_parent_id.setdefault(intf.parent.id, []).append(intf)
        if intf.lag:
            member_intf_by_lag_id.setdefault(intf.lag.id, []).append(intf)

    # fetch IP addresses if requested (one bulk call keyed by assigned_object_id)
    if ip_addresses and devices_to_fetch:
        job.event(f"fetching IP addresses for {len(devices_to_fetch)} device(s)")
        for ip in nb.ipam.ip_addresses.filter(device=devices_to_fetch):
            if (
                ip.assigned_object_id
                and ip.assigned_object_type == "dcim.interface"
            ):
                ip_by_intf_id.setdefault(ip.assigned_object_id, []).append(dict(ip))

    # fetch inventory items if requested (one bulk call keyed by component_id)
    if inventory_items and devices_to_fetch:
        job.event(f"fetching inventory items for {len(devices_to_fetch)} device(s)")
        for item in nb.dcim.inventory_items.filter(device=devices_to_fetch):
            if item.component_id and item.component_type == "dcim.interface":
                inv_by_intf_id.setdefault(item.component_id, []).append(dict(item))

    # transform pynetbox records into result dict keyed by device / interface name
    for intf in all_interfaces:
        device_name = intf.device.name
        if device_name not in ret.result:  # Netbox issue #16299
            continue

        intf_data = dict(intf)
        # add extra fields
        intf_data["child_interfaces"] = [
            {
                "name": c.name,
                "vrf": c.vrf.name if c.vrf else None,
                "ip_addresses": ip_by_intf_id.get(c.id, []),
            }
            for c in children_by_parent_id.get(intf.id, [])
        ]
        intf_data["member_interfaces"] = [
            m.name for m in member_intf_by_lag_id.get(intf.id, [])
        ]
        intf_data["ip_addresses"] = ip_by_intf_id.get(intf.id, [])
        intf_data["inventory_items"] = inv_by_intf_id.get(intf.id, [])
        ret.result[device_name][intf.name] = intf_data

    # cache freshly fetched interfaces per device
    if cache != False:
        job.event(f"caching interfaces data for {len(devices_to_fetch)} device(s)")
        for device_name in devices_to_fetch:
            if device_name in ret.result:
                cache_key = f"get_interfaces::{device_name}"
                self.cache.set(
                    cache_key,
                    ret.result[device_name],
                    expire=self.cache_ttl,
                )

    return ret