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

Retrieves device data from Netbox using the GraphQL API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
filters	list	A list of filter dictionaries to filter devices.	None
instance	str	The Netbox instance name.	None
dry_run	bool	If True, only returns the query content without executing it. Defaults to False.	False
devices	list	A list of device names to query data for.	None
cache	Union[bool, str]	Cache usage options: True: Use data stored in cache if it is up to date, refresh it otherwise. False: Do not use cache and do not update cache. "refresh": Ignore data in cache and replace it with data fetched from Netbox. "force": Use data in cache without checking if it is up to date.	None

Returns:

Name	Type	Description
dict	Result	A dictionary keyed by device name with device data.

Raises:

Type	Description
Exception	If the GraphQL query fails or if there are errors in the query result.

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"]})
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[bool, str] = None,
) -> Result:
    """
    Retrieves device data from Netbox using the GraphQL API.

    Args:
        job: NorFab Job object containing relevant metadata
        filters (list, optional): A list of filter dictionaries to filter devices.
        instance (str, optional): The Netbox instance name.
        dry_run (bool, optional): If True, only returns the query content without executing it. Defaults to False.
        devices (list, optional): A list of device names to query data for.
        cache (Union[bool, str], optional): Cache usage options:

            - True: Use data stored in cache if it is up to date, refresh it otherwise.
            - False: Do not use cache and do not update cache.
            - "refresh": Ignore data in cache and replace it with data fetched from Netbox.
            - "force": Use data in cache without checking if it is up to date.

    Returns:
        dict: A dictionary keyed by device name with device data.

    Raises:
        Exception: If the GraphQL query fails or if there are errors in the query result.
    """
    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 = filters or []
    devices = devices or []
    queries = {}  # devices queries
    device_fields = [
        "name",
        "last_updated",
        "custom_field_data",
        "tags {name}",
        "device_type {model}",
        "role {name}",
        "config_context",
        "tenant {name}",
        "platform {name}",
        "serial",
        "asset_tag",
        "site {name slug tags{name} }",
        "location {name}",
        "rack {name}",
        "status",
        "primary_ip4 {address}",
        "primary_ip6 {address}",
        "airflow",
        "position",
    ]

    if cache == True or cache == "force":
        # retrieve last updated data from Netbox for devices
        last_updated_query = {
            f"devices_by_filter_{index}": {
                "obj": "device_list",
                "filters": filter_item,
                "fields": ["name", "last_updated"],
            }
            for index, filter_item in enumerate(filters)
        }
        if devices:
            # use cache data without checking if it is up to date for cached devices
            if cache == "force":
                for device_name in list(devices):
                    device_cache_key = f"get_devices::{device_name}"
                    if device_cache_key in self.cache:
                        devices.remove(device_name)
                        ret.result[device_name] = self.cache[device_cache_key]
            # query netbox last updated data for devices
            if self.nb_version[instance][0] == 4:
                dlist = '["{dl}"]'.format(dl='", "'.join(devices))
                filters_dict = {"name": f"{{in_list: {dlist}}}"}
            elif self.nb_version[instance][0] == 3:
                filters_dict = {"name": devices}
            last_updated_query["devices_by_devices_list"] = {
                "obj": "device_list",
                "filters": filters_dict,
                "fields": ["name", "last_updated"],
            }
        last_updated = self.graphql(
            job=job,
            queries=last_updated_query,
            instance=instance,
            dry_run=dry_run,
        )
        last_updated.raise_for_status(f"{self.name} - get devices query failed")

        # return dry run result
        if dry_run:
            ret.result["get_devices_dry_run"] = last_updated.result
            return ret

        # try to retrieve device data from cache
        self.cache.expire()  # remove expired items from cache
        for devices_list in last_updated.result.values():
            for device in devices_list:
                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]["last_updated"]
                    == device["last_updated"]
                    or cache == "force"
                ):
                    ret.result[device["name"]] = self.cache[device_cache_key]
                    # remove device from list of devices to retrieve
                    if device["name"] in devices:
                        devices.remove(device["name"])
                # cache old or no cache, fetch device data
                elif device["name"] not in devices:
                    devices.append(device["name"])
    # ignore cache data, fetch data from netbox
    elif cache == False or cache == "refresh":
        queries = {
            f"devices_by_filter_{index}": {
                "obj": "device_list",
                "filters": filter_item,
                "fields": device_fields,
            }
            for index, filter_item in enumerate(filters)
        }

    # fetch devices data from Netbox
    if devices or queries:
        if devices:
            if self.nb_version[instance][0] == 4:
                dlist = '["{dl}"]'.format(dl='", "'.join(devices))
                filters_dict = {"name": f"{{in_list: {dlist}}}"}
            elif self.nb_version[instance][0] == 3:
                filters_dict = {"name": devices}
            queries["devices_by_devices_list"] = {
                "obj": "device_list",
                "filters": filters_dict,
                "fields": device_fields,
            }

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

        # check for errors
        if query_result.errors:
            msg = f"{self.name} - get devices query failed with errors:\n{query_result.errors}"
            raise Exception(msg)

        # return dry run result
        if dry_run:
            ret.result["get_devices_dry_run"] = query_result.result
            return ret

        # process devices data
        devices_data = query_result.result
        for devices_list in devices_data.values():
            for device in devices_list:
                if device["name"] not in ret.result:
                    device_name = device.pop("name")
                    # cache device data
                    if cache != False:
                        cache_key = f"get_devices::{device_name}"
                        self.cache.set(cache_key, device, expire=self.cache_ttl)
                    # add device data to return result
                    ret.result[device_name] = device

    return ret