NetBox Get Devices Task

task api name: get_devices

Retrieves device records from NetBox using the REST API. Results are keyed by device name and include the full device payload returned by pynetbox, with site data expanded for each device.

Inputs

Parameter	Required	Description
filters	No	List of NetBox device filter dictionaries
devices	No	Device names to retrieve
instance	No	NetBox instance name to target
dry_run	No	Return the merged filters without querying NetBox
cache	No	Cache usage mode: True, False, refresh, or force

Output

Normal mode returns a dictionary keyed by device name:

{
    "ceos-leaf-1": {
        "name": "ceos-leaf-1",
        "status": {"value": "active", "label": "Active"},
        "site": {"name": "lab", "...": "..."},
        "platform": {"name": "eos"},
        "primary_ip4": {"address": "192.0.2.11/32"},
        "...": "...",
    },
}

Notes / Gotchas

• When both devices and filters are provided, device names are merged into the filter list as a NetBox name filter.
• cache=True uses cached device data when last_updated matches NetBox. cache="force" uses cached data without the freshness check.
• cache=False skips cache reads and writes. cache="refresh" fetches fresh data and overwrites cache.

Examples

CLIPython

Get specific devices:

nf#netbox get devices devices ceos-leaf-1 ceos-leaf-2

Filter devices using a NetBox filter dictionary:

nf#netbox get devices filters '[{"site": "lab", "status": "active"}]'

Preview the merged filters without querying NetBox:

nf#netbox get devices devices ceos-leaf-1 dry-run

Refresh cached device data:

nf#netbox get devices devices ceos-leaf-1 cache refresh

Context manager:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1", "ceos-leaf-2"],
        },
    )

Direct lifecycle:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "filters": [{"site": "lab", "status": "active"}],
        },
    )

    preview = client.run_job(
        "netbox",
        "get_devices",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1"],
            "dry_run": True,
        },
    )
finally:
    nf.destroy()

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(
    input=GetDevicesInput,
    output=GetDevicesResult,
    fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Get Devices",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
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 is True:
        ret.result["get_devices_dry_run"] = {"filters": filters}
        ret.dry_run = True
        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 = self.bulk_filter(
                nb.dcim.devices,
                **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 self.bulk_filter(nb.dcim.devices, **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