Netbox Get Devices Task

task api name: get_devices

Get Devices Sample Usage

NORFAB Netbox Get Devices Command Shell Reference

NorFab shell supports these command options for Netbox get_devices task:

nf#man tree netbox.get.devices
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── devices:    Query Netbox devices data
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── filters:    List of device filters dictionaries as a JSON string, examples: [{"q": "ceos1"}]
            ├── devices:    Device names to query data for
            ├── dry-run:    Only return query content, do not run it
            └── cache:    How to use cache, default 'True'
nf#

Python API Reference

Retrieve device data from Netbox REST API using Pynetbox.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object	required
filters	Union[None, list]	list of filter dicts applied to dcim/devices/ endpoint, e.g. [{"site": "NYC", "status": "active"}]	None
instance	Union[None, str]	Netbox instance name, uses default if omitted	None
dry_run	bool	if True returns filter params without making REST calls	False
devices	Union[None, list]	list of device names to fetch, merged into filters as {"name": devices}	None
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:

Type	Description
Result	dict keyed by device name with fields: last_updated, custom_field_data, tags, device_type,
Result	role, config_context, tenant, platform, serial, asset_tag, site, location, rack, status,
Result	primary_ip4, primary_ip6, airflow, position, id

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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_devices(
    self,
    job: Job,
    filters: Union[None, list] = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    devices: Union[None, list] = None,
    cache: Union[None, bool, str] = None,
) -> Result:
    """
    Retrieve device data from Netbox REST API using Pynetbox.

    Args:
        job: NorFab Job object
        filters: list of filter dicts applied to ``dcim/devices/`` endpoint, e.g. ``[{"site": "NYC", "status": "active"}]``
        instance: Netbox instance name, uses default if omitted
        dry_run: if True returns filter params without making REST calls
        devices: list of device names to fetch, merged into filters as ``{"name": devices}``
        cache: ``True`` - use cache if up to date; ``False`` - skip cache;
            ``"refresh"`` - fetch and overwrite cache; ``"force"`` - use cache without staleness check

    Returns:
        dict keyed by device name with fields: last_updated, custom_field_data, tags, device_type,
        role, config_context, tenant, platform, serial, asset_tag, site, location, rack, status,
        primary_ip4, primary_ip6, airflow, position, id
    """
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:get_devices", result={}, resources=[instance])
    cache = self.cache_use if cache is None else cache
    filters = list(filters) if filters else []
    devices = devices or []
    devices_to_fetch = []
    sites_data = {}
    nb = self._get_pynetbox(instance)

    # merge named devices into filters as a name filter
    if devices:
        filters.append({"name": devices})

    # return dry run result
    if dry_run:
        ret.result["get_devices_dry_run"] = {"filters": filters}
        return ret

    job.event(
        f"retrieving device data for {len(devices)} device(s) from instance '{instance}'"
        if devices
        else f"retrieving device data from instance '{instance}' using {len(filters)} filter(s)"
    )

    filters_to_fetch = list(filters)

    if cache == True or cache == "force":
        job.event("checking cache for up-to-date device data")
        self.cache.expire()  # remove expired items from cache
        # retrieve last_updated data from Netbox for all filters using REST
        for filter_item in filters:
            result = nb.dcim.devices.filter(
                **filter_item,
                fields="name,last_updated",
            )
            for device in result:
                device_name = device.name
                last_updated = device.last_updated
                # try to retrieve device data from cache
                device_cache_key = f"get_devices::{device_name}"
                # check if cache is up to date and use it if so
                if device_cache_key in self.cache and (
                    self.cache[device_cache_key].get("last_updated") == last_updated
                    or cache == "force"
                ):
                    ret.result[device_name] = self.cache[device_cache_key]
                    job.event(f"serving '{device_name}' from cache")
                # cache old or no cache, fetch device data
                else:
                    devices_to_fetch.append(device_name)
                    job.event(
                        f"'{device_name}' cache miss or stale, fetching fresh data"
                    )

        # only fetch devices missing from or stale in cache
        filters_to_fetch = [{"name": devices_to_fetch}] if devices_to_fetch else []
    # ignore cache, fetch data from Netbox
    elif cache == False or cache == "refresh":
        pass  # filters_to_fetch already set to all filters above

    # fetch full device data from Netbox
    if filters_to_fetch:
        job.event(f"fetching device data from NetBox instance '{instance}'")
        nb = self._get_pynetbox(instance)
        all_devices_raw = {}

        for filter_item in filters_to_fetch:
            for device in nb.dcim.devices.filter(**filter_item):
                all_devices_raw.setdefault(device.name, device)

        job.event(f"retrieved {len(all_devices_raw)} device(s) from NetBox")

        # process devices data
        for device_name, device in all_devices_raw.items():
            if device_name not in ret.result:
                device_data = dict(device)
                if device.site.name not in sites_data:
                    sites_data[device.site.name] = dict(
                        nb.dcim.sites.get(id=device.site.id)
                    )
                device_data["site"] = sites_data[device.site.name]
                # cache device data
                if cache != False:
                    cache_key = f"get_devices::{device_name}"
                    self.cache.set(cache_key, device_data, expire=self.cache_ttl)
                    log.info(
                        f"{self.name} - Cached device data for '{device_name}'"
                    )
                    job.event(f"cached device data for '{device_name}'")
                # add device data to return result
                ret.result[device_name] = device_data

    log.info(f"{self.name} - get_devices returning {len(ret.result)} device(s)")
    job.event(f"fetched {len(ret.result)} device(s)")

    return ret