Netbox Worker

NetboxWorker(inventory, broker, worker_name, exit_event=None, init_done_event=None, log_level=None, log_queue: object = None)

Bases: NFPWorker

NetboxWorker class for interacting with Netbox API and managing inventory.

Parameters:

Name	Type	Description	Default
inventory	dict	The inventory data.	required
broker	object	The broker instance.	required
worker_name	str	The name of the worker.	required
exit_event	Event	Event to signal exit.	None
init_done_event	Event	Event to signal initialization completion.	None
log_level	int	Logging level.	None
log_queue	object	Queue for logging.	None

Raises:

Type	Description
AssertionError	If the inventory has no Netbox instances.

Attributes:

Name	Type	Description
default_instance	str	Default Netbox instance name.
inventory	dict	Inventory data.
nb_version	tuple	Netbox version.
compatible_ge_v3	tuple	Minimum supported Netbox v3 version.
compatible_ge_v4	tuple	Minimum supported Netbox v4 version.

Source code in norfab\workers\netbox_worker.py

def __init__(
    self,
    inventory,
    broker,
    worker_name,
    exit_event=None,
    init_done_event=None,
    log_level=None,
    log_queue: object = None,
):
    super().__init__(
        inventory, broker, SERVICE, worker_name, exit_event, log_level, log_queue
    )
    self.init_done_event = init_done_event
    self.cache = None

    # get inventory from broker
    self.netbox_inventory = self.load_inventory()
    if not self.netbox_inventory:
        log.critical(
            f"{self.name} - Broker {self.broker} returned no inventory for {self.name}, killing myself..."
        )
        self.destroy()

    assert self.netbox_inventory.get(
        "instances"
    ), f"{self.name} - inventory has no Netbox instances"

    # extract parameters from imvemtory
    self.netbox_connect_timeout = self.netbox_inventory.get(
        "netbox_connect_timeout", 10
    )
    self.netbox_read_timeout = self.netbox_inventory.get("netbox_read_timeout", 300)
    self.cache_use = self.netbox_inventory.get("cache_use", True)
    self.cache_ttl = self.netbox_inventory.get("cache_ttl", 31557600)  # 1 Year

    # find default instance
    for name, params in self.netbox_inventory["instances"].items():
        if params.get("default") is True:
            self.default_instance = name
            break
    else:
        self.default_instance = name

    # check Netbox compatibility
    self._verify_compatibility()

    # instantiate cache
    self.cache_dir = os.path.join(self.base_dir, "cache")
    os.makedirs(self.cache_dir, exist_ok=True)
    self.cache = self._get_diskcache()

    self.init_done_event.set()
    log.info(f"{self.name} - Started")

worker_exit() -> None

Worker exist sanity checks. Closes the cache if it exists.

This method checks if the cache attribute is present and not None. If the cache exists, it closes the cache to release any resources associated with it.

Source code in norfab\workers\netbox_worker.py

def worker_exit(self) -> None:
    """
    Worker exist sanity checks. Closes the cache if it exists.

    This method checks if the cache attribute is present and not None.
    If the cache exists, it closes the cache to release any resources
    associated with it.
    """
    if self.cache:
        self.cache.close()

get_inventory() -> dict

NorFab Task to return running inventory for NetBox worker.

Returns:

Name	Type	Description
dict	dict	A dictionary containing the NetBox inventory.

Source code in norfab\workers\netbox_worker.py

def get_inventory(self) -> dict:
    """
    NorFab Task to return running inventory for NetBox worker.

    Returns:
        dict: A dictionary containing the NetBox inventory.
    """
    return Result(
        task=f"{self.name}:get_inventory", result=dict(self.netbox_inventory)
    )

get_version(**kwargs) -> dict

Retrieves the version information of specified libraries and system details.

Returns:

Name	Type	Description
dict	dict	A dictionary containing the version information of the specified libraries and system details.

Source code in norfab\workers\netbox_worker.py

def get_version(self, **kwargs) -> dict:
    """
    Retrieves the version information of specified libraries and system details.

    Returns:
        dict: A dictionary containing the version information of the
            specified libraries and system details.
    """
    libs = {
        "norfab": "",
        "pynetbox": "",
        "requests": "",
        "python": sys.version.split(" ")[0],
        "platform": sys.platform,
        "netbox_version": self.nb_version,
    }
    # get version of packages installed
    for pkg in libs.keys():
        try:
            libs[pkg] = importlib.metadata.version(pkg)
        except importlib.metadata.PackageNotFoundError:
            pass

    return Result(task=f"{self.name}:get_version", result=libs)

get_netbox_status(instance=None) -> dict

Retrieve the status of NetBox instances.

This method queries the status of a specific NetBox instance if the instance parameter is provided. If no instance is specified, it queries the status of all instances in the NetBox inventory.

Parameters:

Name	Type	Description	Default
instance	str	The name of the specific NetBox instance to query.	None

Returns:

Name	Type	Description
dict	dict	A dictionary containing the status of the requested NetBox instance(s).

Source code in norfab\workers\netbox_worker.py

def get_netbox_status(self, instance=None) -> dict:
    """
    Retrieve the status of NetBox instances.

    This method queries the status of a specific NetBox instance if the
    `instance` parameter is provided. If no instance is specified, it
    queries the status of all instances in the NetBox inventory.

    Args:
        instance (str, optional): The name of the specific NetBox instance
                                  to query.

    Returns:
        dict: A dictionary containing the status of the requested NetBox
              instance(s).
    """
    ret = Result(result={}, task=f"{self.name}:get_netbox_status")
    if instance:
        ret.result[instance] = self._query_netbox_status(instance)
    else:
        for name in self.netbox_inventory["instances"].keys():
            ret.result[name] = self._query_netbox_status(name)
    return ret

get_compatibility() -> dict

Checks the compatibility of Netbox instances based on their version.

This method retrieves the status and version of Netbox instances and determines if they are compatible with the required versions. It logs a warning if any instance is not reachable.

Returns:

Name	Type	Description
dict	dict	A dictionary where the keys are the instance names and the values are booleans indicating compatibility (True/False) or None if the instance is not reachable.

Source code in norfab\workers\netbox_worker.py

def get_compatibility(self) -> dict:
    """
    Checks the compatibility of Netbox instances based on their version.

    This method retrieves the status and version of Netbox instances and determines
    if they are compatible with the required versions. It logs a warning if any
    instance is not reachable.

    Returns:
        dict: A dictionary where the keys are the instance names and the values are
              booleans indicating compatibility (True/False) or None if the instance
              is not reachable.
    """
    ret = Result(task=f"{self.name}:get_compatibility", result={})
    netbox_status = self.get_netbox_status()
    for instance, params in netbox_status.result.items():
        if params["status"] is not True:
            log.warning(f"{self.name} - {instance} Netbox instance not reachable")
            ret.result[instance] = None
        else:
            if "-docker-" in params["netbox-version"].lower():
                self.nb_version[instance] = tuple(
                    [
                        int(i)
                        for i in params["netbox-version"]
                        .lower()
                        .split("-docker-")[0]
                        .split(".")
                    ]
                )
            else:
                self.nb_version[instance] = tuple(
                    [int(i) for i in params["netbox-version"].split(".")]
                )
            if self.nb_version[instance][0] == 3:  # check Netbox 3 compatibility
                ret.result[instance] = (
                    self.nb_version[instance] >= self.compatible_ge_v3
                )
            elif self.nb_version[instance][0] == 4:  # check Netbox 4 compatibility
                ret.result[instance] = (
                    self.nb_version[instance] >= self.compatible_ge_v4
                )

    return ret

cache_list(keys='*', details=False) -> list

Retrieve a list of cache keys, optionally with details about each key.

Parameters:

Name	Type	Description	Default
keys	str	A pattern to match cache keys against. Defaults to "*".	'*'
details	bool	If True, include detailed information about each cache key. Defaults to False.	False

Returns:

Name	Type	Description
list	list	A list of cache keys or a list of dictionaries with detailed information if details is True.

Source code in norfab\workers\netbox_worker.py

def cache_list(self, keys="*", details=False) -> list:
    """
    Retrieve a list of cache keys, optionally with details about each key.

    Args:
        keys (str): A pattern to match cache keys against. Defaults to "*".
        details (bool): If True, include detailed information about each cache key. Defaults to False.

    Returns:
        list: A list of cache keys or a list of dictionaries with detailed information if `details` is True.
    """
    self.cache.expire()
    ret = Result(task=f"{self.name}:cache_list", result=[])
    for cache_key in self.cache:
        if fnmatchcase(cache_key, keys):
            if details:
                _, expires = self.cache.get(cache_key, expire_time=True)
                expires = datetime.fromtimestamp(expires)
                creation = expires - timedelta(seconds=self.cache_ttl)
                age = datetime.now() - creation
                ret.result.append(
                    {
                        "key": cache_key,
                        "age": str(age),
                        "creation": str(creation),
                        "expires": str(expires),
                    }
                )
            else:
                ret.result.append(cache_key)
    return ret

cache_clear(key=None, keys=None) -> list

Clears specified keys from the cache.

Parameters:

Name	Type	Description	Default
key	str	A specific key to remove from the cache.	None
keys	str	A glob pattern to match multiple keys to remove from the cache.	None

Returns:

Name	Type	Description
list	list	A list of keys that were successfully removed from the cache.

Raises:

Type	Description
RuntimeError	If a specified key or a key matching the glob pattern could not be removed from the cache.

Notes:

• If neither key nor keys is provided, the function will return a message indicating that there is nothing to clear.
• If key is provided, it will attempt to remove that specific key from the cache.
• If keys is provided, it will attempt to remove all keys matching the glob pattern from the cache.

Source code in norfab\workers\netbox_worker.py

def cache_clear(self, key=None, keys=None) -> list:
    """
    Clears specified keys from the cache.

    Args:
        key (str, optional): A specific key to remove from the cache.
        keys (str, optional): A glob pattern to match multiple keys to remove from the cache.

    Returns:
        list: A list of keys that were successfully removed from the cache.

    Raises:
        RuntimeError: If a specified key or a key matching the glob pattern could not be removed from the cache.

    Notes:

    - If neither `key` nor `keys` is provided, the function will return a message indicating that there is nothing to clear.
    - If `key` is provided, it will attempt to remove that specific key from the cache.
    - If `keys` is provided, it will attempt to remove all keys matching the glob pattern from the cache.
    """
    ret = Result(task=f"{self.name}:cache_clear", result=[])
    # check if has keys to clear
    if key == keys == None:
        ret.result = "Noting to clear, specify key or keys"
        return ret
    # remove specific key from cache
    if key:
        if key in self.cache:
            if self.cache.delete(key, retry=True):
                ret.result.append(key)
            else:
                raise RuntimeError(f"Failed to remove {key} from cache")
        else:
            ret.messages.append(f"Key {key} not in cache.")
    # remove all keys matching glob pattern
    if keys:
        for cache_key in self.cache:
            if fnmatchcase(cache_key, keys):
                if self.cache.delete(cache_key, retry=True):
                    ret.result.append(cache_key)
                else:
                    raise RuntimeError(f"Failed to remove {key} from cache")
    return ret

cache_get(key=None, keys=None, raise_missing=False) -> dict

Retrieve values from the cache based on a specific key or a pattern of keys.

Parameters:

Name	Type	Description	Default
key	str	A specific key to retrieve from the cache.	None
keys	str	A glob pattern to match multiple keys in the cache.	None
raise_missing	bool	If True, raises a KeyError if the specific key is not found in the cache. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	A dictionary containing the results of the cache retrieval. The keys are the cache keys and the values are the corresponding cache values.

Raises:

Type	Description
KeyError	If raise_missing is True and the specific key is not found in the cache.

Source code in norfab\workers\netbox_worker.py

def cache_get(self, key=None, keys=None, raise_missing=False) -> dict:
    """
    Retrieve values from the cache based on a specific key or a pattern of keys.

    Args:
        key (str, optional): A specific key to retrieve from the cache.
        keys (str, optional): A glob pattern to match multiple keys in the cache.
        raise_missing (bool, optional): If True, raises a KeyError if the specific
            key is not found in the cache. Defaults to False.

    Returns:
        dict: A dictionary containing the results of the cache retrieval. The keys are
            the cache keys and the values are the corresponding cache values.

    Raises:
        KeyError: If raise_missing is True and the specific key is not found in the cache.
    """
    ret = Result(task=f"{self.name}:cache_clear", result={})
    # get specific key from cache
    if key:
        if key in self.cache:
            ret.result[key] = self.cache[key]
        elif raise_missing:
            raise KeyError(f"Key {key} not in cache.")
    # get all keys matching glob pattern
    if keys:
        for cache_key in self.cache:
            if fnmatchcase(cache_key, keys):
                ret.result[cache_key] = self.cache[cache_key]
    return ret

graphql(instance: str = None, dry_run: bool = False, obj: dict = None, filters: dict = None, fields: list = None, queries: dict = None, query_string: str = None) -> Union[dict, list]

Function to query Netbox v3 or Netbox v4 GraphQL API.

Parameters:

Name	Type	Description	Default
instance	str	Netbox instance name	None
dry_run	bool	only return query content, do not run it	False
obj	dict	Object to query	None
filters	dict	Filters to apply to the query	None
fields	list	Fields to retrieve in the query	None
queries	dict	Dictionary of queries to execute	None
query_string	str	Raw query string to execute	None

Returns:

Name	Type	Description
dict	Union[dict, list]	GraphQL request data returned by Netbox

Raises:

Type	Description
RuntimeError	If required arguments are not provided
Exception	If GraphQL query fails

Source code in norfab\workers\netbox_worker.py

def graphql(
    self,
    instance: str = None,
    dry_run: bool = False,
    obj: dict = None,
    filters: dict = None,
    fields: list = None,
    queries: dict = None,
    query_string: str = None,
) -> Union[dict, list]:
    """
    Function to query Netbox v3 or Netbox v4 GraphQL API.

    Args:
        instance: Netbox instance name
        dry_run: only return query content, do not run it
        obj: Object to query
        filters: Filters to apply to the query
        fields: Fields to retrieve in the query
        queries: Dictionary of queries to execute
        query_string: Raw query string to execute

    Returns:
        dict: GraphQL request data returned by Netbox

    Raises:
        RuntimeError: If required arguments are not provided
        Exception: If GraphQL query fails
    """
    nb_params = self._get_instance_params(instance)
    ret = Result(task=f"{self.name}:graphql")
    instance = instance or self.default_instance

    # form graphql query(ies) payload
    if queries:
        queries_list = []
        for alias, query_data in queries.items():
            query_data["alias"] = alias
            if self.nb_version[instance][0] == 4:
                queries_list.append(_form_query_v4(**query_data))
            elif self.nb_version[instance][0] == 3:
                queries_list.append(_form_query_v3(**query_data))
        queries_strings = "    ".join(queries_list)
        query = f"query {{{queries_strings}}}"
    elif obj and filters and fields:
        if self.nb_version[instance][0] == 4:
            query = _form_query_v4(obj, filters, fields)
        elif self.nb_version[instance][0] == 3:
            query = _form_query_v3(obj, filters, fields)
        query = f"query {{{query}}}"
    elif query_string:
        query = query_string
    else:
        raise RuntimeError(
            f"{self.name} - graphql method expects quieries argument or obj, filters, "
            f"fields arguments or query_string argument provided"
        )
    payload = json.dumps({"query": query})

    # form and return dry run response
    if dry_run:
        ret.result = {
            "url": f"{nb_params['url']}/graphql/",
            "data": payload,
            "verify": nb_params.get("ssl_verify", True),
            "headers": {
                "Content-Type": "application/json",
                "Accept": "application/json",
                "Authorization": f"Token ...{nb_params['token'][-6:]}",
            },
        }
        return ret

    # send request to Netbox GraphQL API
    log.debug(
        f"{self.name} - sending GraphQL query '{payload}' to URL '{nb_params['url']}/graphql/'"
    )
    req = requests.post(
        url=f"{nb_params['url']}/graphql/",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Token {nb_params['token']}",
        },
        data=payload,
        verify=nb_params.get("ssl_verify", True),
        timeout=(self.netbox_connect_timeout, self.netbox_read_timeout),
    )
    try:
        req.raise_for_status()
    except Exception as e:
        raise Exception(
            f"{self.name} -  Netbox GraphQL query failed, query '{query}', "
            f"URL '{req.url}', status-code '{req.status_code}', reason '{req.reason}', "
            f"response content '{req.text}'"
        )

    # return results
    reply = req.json()
    if reply.get("errors"):
        msg = f"{self.name} - GrapQL query error '{reply['errors']}', query '{payload}'"
        log.error(msg)
        ret.errors.append(msg)
        if reply.get("data"):
            ret.result = reply["data"]  # at least return some data
    elif queries or query_string:
        ret.result = reply["data"]
    else:
        ret.result = reply["data"][obj]

    return ret

rest(instance: str = None, method: str = 'get', api: str = '', **kwargs) -> Union[dict, list]

Sends a request to the Netbox REST API.

Parameters:

Name	Type	Description	Default
instance	str	The instance name to get parameters for.	None
method	str	The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".	'get'
api	str	The API endpoint to send the request to. Defaults to "".	''
**kwargs		Additional arguments to pass to the request (e.g., params, data, json).	{}

Returns:

Type	Description
Union[dict, list]	Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

Raises:

Type	Description
HTTPError	If the HTTP request returned an unsuccessful status code.

Source code in norfab\workers\netbox_worker.py

def rest(
    self, instance: str = None, method: str = "get", api: str = "", **kwargs
) -> Union[dict, list]:
    """
    Sends a request to the Netbox REST API.

    Args:
        instance (str, optional): The instance name to get parameters for.
        method (str, optional): The HTTP method to use for the request (e.g., 'get', 'post'). Defaults to "get".
        api (str, optional): The API endpoint to send the request to. Defaults to "".
        **kwargs: Additional arguments to pass to the request (e.g., params, data, json).

    Returns:
        Union[dict, list]: The JSON response from the API, parsed into a dictionary or list.

    Raises:
        requests.exceptions.HTTPError: If the HTTP request returned an unsuccessful status code.
    """
    params = self._get_instance_params(instance)

    # send request to Netbox REST API
    response = getattr(requests, method)(
        url=f"{params['url']}/api/{api}/",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Token {params['token']}",
        },
        verify=params.get("ssl_verify", True),
        **kwargs,
    )

    response.raise_for_status()

    return response.json()

get_devices(filters: list = None, instance: str = None, dry_run: bool = False, devices: list = None, cache: Union[bool, str] = None) -> dict

Retrieves device data from Netbox using the GraphQL API.

Parameters:

Name	Type	Description	Default
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	dict	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

def get_devices(
    self,
    filters: list = None,
    instance: str = None,
    dry_run: bool = False,
    devices: list = None,
    cache: Union[bool, str] = None,
) -> dict:
    """
    Retrieves device data from Netbox using the GraphQL API.

    Args:
        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.
    """
    ret = Result(task=f"{self.name}:get_devices", result={})
    cache = self.cache_use if cache is None else cache
    instance = instance or self.default_instance
    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(
            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(
            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

get_interfaces(instance: str = None, devices: list = None, ip_addresses: bool = False, inventory_items: bool = False, dry_run: bool = False, cache: Union[bool, str] = None) -> dict

Retrieve device interfaces from Netbox using GraphQL API.

Parameters:

Name	Type	Description	Default
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	dict	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

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

    Args:
        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.
    """
    # form final result object
    ret = Result(
        task=f"{self.name}:get_interfaces", result={d: {} for d in devices}
    )
    instance = instance or self.default_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
    queries = {
        "interfaces": {
            "obj": "interface_list",
            "filters": {"device": devices},
            "fields": intf_fields,
        }
    }

    # add query to retrieve inventory items
    if inventory_items:
        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(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

get_connections(devices: list, instance: str = None, dry_run: bool = False, cables: bool = False, cache: Union[bool, str] = None) -> dict

Retrieve interface connection details for specified devices from Netbox.

Parameters:

Name	Type	Description	Default
devices	list	List of device names to retrieve connections for.	required
instance	str	Instance name for the GraphQL query.	None
dry_run	bool	If True, perform a dry run without making actual changes.	False
cables	bool	if True includes interfaces' directly attached cables details	False

Returns:

Name	Type	Description
dict	dict	A dictionary containing connection details for each device: { "netbox-worker-1.2": { "r1": { "Console": { "breakout": false, "remote_device": "termserv1", "remote_interface": "ConsoleServerPort1", "remote_termination_type": "consoleserverport", "termination_type": "consoleport" }, "eth1": { "breakout": false, "remote_device": "r2", "remote_interface": "eth8", "remote_termination_type": "interface", "termination_type": "interface" } } } }

Raises:

Type	Description
Exception	If there is an error in the GraphQL query or data retrieval process.

Source code in norfab\workers\netbox_worker.py

def get_connections(
    self,
    devices: list,
    instance: str = None,
    dry_run: bool = False,
    cables: bool = False,
    cache: Union[bool, str] = None,
) -> dict:
    """
    Retrieve interface connection details for specified devices from Netbox.

    Args:
        devices (list): List of device names to retrieve connections for.
        instance (str, optional): Instance name for the GraphQL query.
        dry_run (bool, optional): If True, perform a dry run without making actual changes.
        cables (bool, optional): if True includes interfaces' directly attached cables details

    Returns:
        dict: A dictionary containing connection details for each device:

            ```
            {
                "netbox-worker-1.2": {
                    "r1": {
                        "Console": {
                            "breakout": false,
                            "remote_device": "termserv1",
                            "remote_interface": "ConsoleServerPort1",
                            "remote_termination_type": "consoleserverport",
                            "termination_type": "consoleport"
                        },
                        "eth1": {
                            "breakout": false,
                            "remote_device": "r2",
                            "remote_interface": "eth8",
                            "remote_termination_type": "interface",
                            "termination_type": "interface"
                        }
                    }
                }
            }
            ```

    Raises:
        Exception: If there is an error in the GraphQL query or data retrieval process.
    """
    # form final result dictionary
    ret = Result(
        task=f"{self.name}:get_connections", result={d: {} for d in devices}
    )
    instance = instance or self.default_instance

    # form lists of fields to request from netbox
    cable_fields = """
        cable {
            type
            status
            tenant {name}
            label
            tags {name}
            length
            length_unit
            custom_fields
        }
    """
    if self.nb_version[instance][0] == 4:
        interfaces_fields = [
            "name",
            "device {name}",
            """connected_endpoints {
            __typename 
            ... on InterfaceType {name device {name}}
            ... on ProviderNetworkType {name}
            }""",
        ]
    elif self.nb_version[instance][0] == 3:
        interfaces_fields = [
            "name",
            "device {name}",
            """connected_endpoints {
            __typename 
            ... on InterfaceType {name device {name}}
            }""",
        ]
    interfaces_fields.append(
        """
        link_peers {
            __typename
            ... on InterfaceType {name device {name}}
            ... on FrontPortType {name device {name}}
            ... on RearPortType {name device {name}}
        }
    """
    )
    console_ports_fields = [
        "name",
        "device {name}",
        """connected_endpoints {
          __typename 
          ... on ConsoleServerPortType {name device {name}}
        }""",
        """link_peers {
          __typename
          ... on ConsoleServerPortType {name device {name}}
          ... on FrontPortType {name device {name}}
          ... on RearPortType {name device {name}}
        }""",
    ]
    console_server_ports_fields = [
        "name",
        "device {name}",
        """connected_endpoints {
          __typename 
          ... on ConsolePortType {name device {name}}
        }""",
        """link_peers {
          __typename
          ... on ConsolePortType {name device {name}}
          ... on FrontPortType {name device {name}}
          ... on RearPortType {name device {name}}
        }""",
    ]

    # check if need to include cables info
    if cables is True:
        interfaces_fields.append(cable_fields)
        console_ports_fields.append(cable_fields)
        console_server_ports_fields.append(cable_fields)

    # form query dictionary with aliases to get data from Netbox
    queries = {
        "interface": {
            "obj": "interface_list",
            "filters": {"device": devices},
            "fields": interfaces_fields,
        },
        "consoleport": {
            "obj": "console_port_list",
            "filters": {"device": devices},
            "fields": console_ports_fields,
        },
        "consoleserverport": {
            "obj": "console_server_port_list",
            "filters": {"device": devices},
            "fields": console_server_ports_fields,
        },
    }

    # retrieve full list of devices interface with all cables
    query_result = self.graphql(queries=queries, instance=instance, dry_run=dry_run)

    # return dry run result
    if dry_run:
        return query_result

    all_ports = query_result.result

    # extract interfaces
    for port_type, ports in all_ports.items():
        for port in ports:
            endpoints = port["connected_endpoints"]
            # skip ports that have no remote device connected
            if not endpoints or not all(i for i in endpoints):
                continue

            # extract required parameters
            cable = port.get("cable", {})
            device_name = port["device"]["name"]
            port_name = port["name"]
            link_peers = port["link_peers"]
            remote_termination_type = endpoints[0]["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")

            # form initial connection dictionary
            connection = {
                "breakout": len(endpoints) > 1,
                "remote_termination_type": remote_termination_type,
                "termination_type": port_type,
            }

            # add remote connection details
            if remote_termination_type == "providernetwork":
                connection["remote_device"] = None
                connection["remote_interface"] = None
                connection["provider"] = endpoints[0]["name"]
            else:
                remote_interface = endpoints[0]["name"]
                if len(endpoints) > 1:
                    remote_interface = list(sorted([i["name"] for i in endpoints]))
                connection["remote_interface"] = remote_interface
                connection["remote_device"] = endpoints[0]["device"]["name"]

            # add cable and its peer details
            if cables:
                peer_termination_type = link_peers[0]["__typename"].lower()
                peer_termination_type = peer_termination_type.replace("type", "")
                cable["peer_termination_type"] = peer_termination_type
                cable["peer_device"] = link_peers[0].get("device", {}).get("name")
                cable["peer_interface"] = link_peers[0].get("name")
                if len(link_peers) > 1:  # handle breakout cable
                    cable["peer_interface"] = [i["name"] for i in link_peers]
                connection["cable"] = cable

            ret.result[device_name][port_name] = connection

    return ret

get_circuits(devices: list, cid: list = None, instance: str = None, dry_run: bool = False, cache: Union[bool, str] = True) -> dict

Retrieve circuit information for specified devices from Netbox.

Parameters:

Name	Type	Description	Default
devices	list	List of device names to retrieve circuits for.	required
cid	list	List of circuit IDs to filter by.	None
instance	str	Netbox instance to query.	None
dry_run	bool	If True, perform a dry run without making changes. Defaults to False.	False
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.	True

Returns:

Name	Type	Description
dict	dict	dictionary keyed by device names with circuits data.

Task to retrieve device's circuits data from Netbox.

Source code in norfab\workers\netbox_worker.py

def get_circuits(
    self,
    devices: list,
    cid: list = None,
    instance: str = None,
    dry_run: bool = False,
    cache: Union[bool, str] = True,
) -> dict:
    """

    Retrieve circuit information for specified devices from Netbox.

    Args:
        devices (list): List of device names to retrieve circuits for.
        cid (list, optional): List of circuit IDs to filter by.
        instance (str, optional): Netbox instance to query.
        dry_run (bool, optional): If True, perform a dry run without making changes. Defaults to False.
        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: dictionary keyed by device names with circuits data.

    Task to retrieve device's circuits data from Netbox.
    """
    log.info(
        f"{self.name}:get_circuits - {instance or self.default_instance} Netbox, "
        f"devices {', '.join(devices)}, cid {cid}"
    )

    # form final result object
    ret = Result(task=f"{self.name}:get_circuits", result={d: {} for d in devices})
    instance = instance or self.default_instance
    cache = self.cache_use if cache is None else cache
    cid = cid or []
    circuit_fields = [
        "cid",
        "tags {name}",
        "provider {name}",
        "commit_rate",
        "description",
        "status",
        "type {name}",
        "provider_account {name}",
        "tenant {name}",
        "termination_a {id last_updated}",
        "termination_z {id last_updated}",
        "custom_fields",
        "comments",
        "last_updated",
    ]

    # form initial circuits filters based on devices' sites and cid list
    circuits_filters_dict = {}
    device_data = self.get_devices(
        devices=copy.deepcopy(devices), instance=instance, cache=cache
    )
    sites = list(set([i["site"]["slug"] for i in device_data.result.values()]))
    if self.nb_version[instance][0] == 4:
        circuits_filters_dict = {"site": sites}
        if cid:
            cid_list = '["{cl}"]'.format(cl='", "'.join(cid))
            circuits_filters_dict["cid"] = f"{{in_list: {cid_list}}}"
    elif self.nb_version[instance][0] == 3:
        circuits_filters_dict = {"site": sites}
        if cid:
            cid_list = '["{cl}"]'.format(cl='", "'.join(cid))
            circuits_filters_dict["cid"] = cid_list

    log.info(
        f"{self.name}:get_circuits - constructed circuits filters: {circuits_filters_dict}"
    )

    if cache == True or cache == "force":
        log.info(f"{self.name}:get_circuits - retrieving circuits data from cache")
        cid_list = []  #  new cid list for follow up query
        # retrieve last updated data from Netbox for circuits and their terminations
        last_updated = self.graphql(
            obj="circuit_list",
            filters=circuits_filters_dict,
            fields=[
                "cid",
                "last_updated",
                "termination_a {id last_updated}",
                "termination_z {id last_updated}",
            ],
            dry_run=dry_run,
            instance=instance,
        )
        last_updated.raise_for_status(f"{self.name} - get circuits query failed")

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

        # retrieve circuits data from cache
        self.cache.expire()  # remove expired items from cache
        for device in devices:
            for circuit in last_updated.result:
                circuit_cache_key = f"get_circuits::{circuit['cid']}"
                log.info(
                    f"{self.name}:get_circuits - searching cache for key {circuit_cache_key}"
                )
                # check if cache is up to date and use it if so
                if circuit_cache_key in self.cache:
                    cache_ckt = self.cache[circuit_cache_key]
                    # check if device uses this circuit
                    if device not in cache_ckt:
                        continue
                    # use cache forcefully
                    if cache == "force":
                        ret.result[device][circuit["cid"]] = cache_ckt[device]
                    # check circuit cache is up to date
                    if cache_ckt[device]["last_updated"] != circuit["last_updated"]:
                        continue
                    if (
                        cache_ckt[device]["termination_a"]
                        and circuit["termination_a"]
                        and cache_ckt[device]["termination_a"]["last_updated"]
                        != circuit["termination_a"]["last_updated"]
                    ):
                        continue
                    if (
                        cache_ckt[device]["termination_z"]
                        and circuit["termination_z"]
                        and cache_ckt[device]["termination_z"]["last_updated"]
                        != circuit["termination_z"]["last_updated"]
                    ):
                        continue
                    ret.result[device][circuit["cid"]] = cache_ckt[device]
                    log.info(
                        f"{self.name}:get_circuits - {circuit['cid']} retrieved data from cache"
                    )
                elif circuit["cid"] not in cid_list:
                    cid_list.append(circuit["cid"])
                    log.info(
                        f"{self.name}:get_circuits - {circuit['cid']} no cache data found, fetching from Netbox"
                    )
        # form new filters dictionary to fetch remaining circuits data
        circuits_filters_dict = {}
        if cid_list:
            cid_list = '["{cl}"]'.format(cl='", "'.join(cid_list))
            if self.nb_version[instance][0] == 4:
                circuits_filters_dict["cid"] = f"{{in_list: {cid_list}}}"
            elif self.nb_version[instance][0] == 3:
                circuits_filters_dict["cid"] = cid_list
    # ignore cache data, fetch circuits from netbox
    elif cache == False or cache == "refresh":
        pass

    if circuits_filters_dict:
        query_result = self.graphql(
            obj="circuit_list",
            filters=circuits_filters_dict,
            fields=circuit_fields,
            dry_run=dry_run,
            instance=instance,
        )
        query_result.raise_for_status(f"{self.name} - get circuits query failed")

        # return dry run result
        if dry_run is True:
            return query_result

        all_circuits = query_result.result

        # iterate over circuits and map them to devices
        log.info(
            f"{self.name}:get_circuits - retrieved data for {len(all_circuits)} "
            f"circuits from netbox, mapping circuits to devices"
        )
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            results = [
                executor.submit(
                    self._map_circuit, circuit, ret, instance, devices, cache
                )
                for circuit in all_circuits
            ]
            for _ in concurrent.futures.as_completed(results):
                continue

    return ret

get_nornir_inventory(filters: list = None, devices: list = None, instance: str = None, interfaces: Union[dict, bool] = False, connections: Union[dict, bool] = False, circuits: Union[dict, bool] = False, nbdata: bool = True, primary_ip: str = 'ip4') -> dict

Retrieve and construct Nornir inventory from NetBox data.

Parameters:

Name	Type	Description	Default
filters	list	List of filters to apply when retrieving devices from NetBox.	None
devices	list	List of specific devices to retrieve from NetBox.	None
instance	str	NetBox instance to use.	None
interfaces	Union[dict, bool]	If True, include interfaces data in the inventory. If a dict, use it as arguments for the get_interfaces method.	False
connections	Union[dict, bool]	If True, include connections data in the inventory. If a dict, use it as arguments for the get_connections method.	False
circuits	Union[dict, bool]	If True, include circuits data in the inventory. If a dict, use it as arguments for the get_circuits method.	False
nbdata	bool	If True, include a copy of NetBox device's data in the host's data.	True
primary_ip	str	Specify whether to use 'ip4' or 'ip6' for the primary IP address. Defaults to 'ip4'.	'ip4'

Returns:

Name	Type	Description
dict	dict	Nornir inventory dictionary containing hosts and their respective data.

Source code in norfab\workers\netbox_worker.py

def get_nornir_inventory(
    self,
    filters: list = None,
    devices: list = None,
    instance: str = None,
    interfaces: Union[dict, bool] = False,
    connections: Union[dict, bool] = False,
    circuits: Union[dict, bool] = False,
    nbdata: bool = True,
    primary_ip: str = "ip4",
) -> dict:
    """
    Retrieve and construct Nornir inventory from NetBox data.

    Args:
        filters (list, optional): List of filters to apply when retrieving devices from NetBox.
        devices (list, optional): List of specific devices to retrieve from NetBox.
        instance (str, optional): NetBox instance to use.
        interfaces (Union[dict, bool], optional): If True, include interfaces data
                in the inventory. If a dict, use it as arguments for the get_interfaces method.
        connections (Union[dict, bool], optional): If True, include connections data
                in the inventory. If a dict, use it as arguments for the get_connections method.
        circuits (Union[dict, bool], optional): If True, include circuits data in the
                inventory. If a dict, use it as arguments for the get_circuits method.
        nbdata (bool, optional): If True, include a copy of NetBox device's data in the host's data.
        primary_ip (str, optional): Specify whether to use 'ip4' or 'ip6' for the primary
                IP address. Defaults to 'ip4'.

    Returns:
        dict: Nornir inventory dictionary containing hosts and their respective data.
    """
    hosts = {}
    inventory = {"hosts": hosts}
    ret = Result(task=f"{self.name}:get_nornir_inventory", result=inventory)

    # check Netbox status
    netbox_status = self.get_netbox_status(instance=instance)
    if netbox_status.result[instance or self.default_instance]["status"] is False:
        return ret

    # retrieve devices data
    nb_devices = self.get_devices(
        filters=filters, devices=devices, instance=instance
    )

    # form Nornir hosts inventory
    for device_name, device in nb_devices.result.items():
        host = device["config_context"].pop("nornir", {})
        host.setdefault("data", {})
        name = host.pop("name", device_name)
        hosts[name] = host
        # add platform if not provided in device config context
        if not host.get("platform"):
            if device["platform"]:
                host["platform"] = device["platform"]["name"]
            else:
                log.warning(f"{self.name} - no platform found for '{name}' device")
        # add hostname if not provided in config context
        if not host.get("hostname"):
            if device["primary_ip4"] and primary_ip in ["ip4", "ipv4"]:
                host["hostname"] = device["primary_ip4"]["address"].split("/")[0]
            elif device["primary_ip6"] and primary_ip in ["ip6", "ipv6"]:
                host["hostname"] = device["primary_ip6"]["address"].split("/")[0]
            else:
                host["hostname"] = name
        # add netbox data to host's data
        if nbdata is True:
            host["data"].update(device)

    # return if no hosts found for provided parameters
    if not hosts:
        log.warning(f"{self.name} - no viable hosts returned by Netbox")
        return ret

    # add interfaces data
    if interfaces:
        # decide on get_interfaces arguments
        kwargs = interfaces if isinstance(interfaces, dict) else {}
        # add 'interfaces' key to all hosts' data
        for host in hosts.values():
            host["data"].setdefault("interfaces", {})
        # query interfaces data from netbox
        nb_interfaces = self.get_interfaces(
            devices=list(hosts), instance=instance, **kwargs
        )
        # save interfaces data to hosts' inventory
        while nb_interfaces.result:
            device, device_interfaces = nb_interfaces.result.popitem()
            hosts[device]["data"]["interfaces"] = device_interfaces

    # add connections data
    if connections:
        # decide on get_interfaces arguments
        kwargs = connections if isinstance(connections, dict) else {}
        # add 'connections' key to all hosts' data
        for host in hosts.values():
            host["data"].setdefault("connections", {})
        # query connections data from netbox
        nb_connections = self.get_connections(
            devices=list(hosts), instance=instance, **kwargs
        )
        # save connections data to hosts' inventory
        while nb_connections.result:
            device, device_connections = nb_connections.result.popitem()
            hosts[device]["data"]["connections"] = device_connections

    # add circuits data
    if circuits:
        # decide on get_interfaces arguments
        kwargs = circuits if isinstance(circuits, dict) else {}
        # add 'circuits' key to all hosts' data
        for host in hosts.values():
            host["data"].setdefault("circuits", {})
        # query circuits data from netbox
        nb_circuits = self.get_circuits(
            devices=list(hosts), instance=instance, **kwargs
        )
        # save circuits data to hosts' inventory
        while nb_circuits.result:
            device, device_circuits = nb_circuits.result.popitem()
            hosts[device]["data"]["circuits"] = device_circuits

    return ret

get_nornir_hosts(kwargs: dict, timeout: int)

Retrieves a list of unique Nornir hosts from Nornir service based on provided filter criteria.

Parameters:

Name	Type	Description	Default
kwargs	dict	Dictionary of keyword arguments, where keys starting with 'F' are used as filters.	required
timeout	int	Timeout value (in seconds) for the job execution.	required

Returns:

Name	Type	Description
list		Sorted list of unique Nornir host names that match the filter criteria.

Notes

• Only filters with keys starting with 'F' are considered.
• Hosts are collected from all workers where the job did not fail.

Source code in norfab\workers\netbox_worker.py

def get_nornir_hosts(self, kwargs: dict, timeout: int):
    """
    Retrieves a list of unique Nornir hosts from Nornir service based on provided filter criteria.

    Args:
        kwargs (dict): Dictionary of keyword arguments, where keys starting with 'F' are used as filters.
        timeout (int): Timeout value (in seconds) for the job execution.

    Returns:
        list: Sorted list of unique Nornir host names that match the filter criteria.

    Notes:
        - Only filters with keys starting with 'F' are considered.
        - Hosts are collected from all workers where the job did not fail.
    """
    ret = []
    filters = {k: v for k, v in kwargs.items() if k.startswith("F")}
    if filters:
        nornir_hosts = self.client.run_job(
            "nornir",
            "get_nornir_hosts",
            kwargs=filters,
            workers="all",
            timeout=timeout,
        )
        for w, r in nornir_hosts.items():
            if r["failed"] is False:
                ret.extend(r["result"])

    return list(sorted(set(ret)))

update_device_facts(instance: str = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: list = None, batch_size: int = 10, **kwargs) -> dict

Updates the device facts in NetBox:

• serial number

Parameters:

Name	Type	Description	Default
instance	str	The NetBox instance to use.	None
dry_run	bool	If True, no changes will be made to NetBox.	False
datasource	str	The data source to use. Supported datasources: nornir - uses Nornir Service parse task to retrieve devices' data using NAPALM get_facts getter	'nornir'
timeout	int	The timeout for the job execution. Defaults to 60.	60
devices	list	The list of devices to update.	None
batch_size	int	The number of devices to process in each batch.	10
**kwargs		Additional keyword arguments to pass to the job.	{}

Returns:

Name	Type	Description
dict	dict	A dictionary containing the results of the update operation.

Raises:

Type	Description
Exception	If a device does not exist in NetBox.
UnsupportedServiceError	If the specified datasource is not supported.

Source code in norfab\workers\netbox_worker.py

def update_device_facts(
    self,
    instance: str = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: list = None,
    batch_size: int = 10,
    **kwargs,
) -> dict:
    """
    Updates the device facts in NetBox:

    - serial number

    Args:
        instance (str, optional): The NetBox instance to use.
        dry_run (bool, optional): If True, no changes will be made to NetBox.
        datasource (str, optional): The data source to use. Supported datasources:

            - **nornir** - uses Nornir Service parse task to retrieve devices' data
                using NAPALM get_facts getter

        timeout (int, optional): The timeout for the job execution. Defaults to 60.
        devices (list, optional): The list of devices to update.
        batch_size (int, optional): The number of devices to process in each batch.
        **kwargs: Additional keyword arguments to pass to the job.

    Returns:
        dict: A dictionary containing the results of the update operation.

    Raises:
        Exception: If a device does not exist in NetBox.
        UnsupportedServiceError: If the specified datasource is not supported.
    """
    result = {}
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:update_device_facts", result=result)
    nb = self._get_pynetbox(instance)
    kwargs["add_details"] = True

    if datasource == "nornir":
        # source hosts list from Nornir
        if kwargs:
            devices.extend(self.get_nornir_hosts(kwargs, timeout))
        # iterate over devices in batches
        for i in range(0, len(devices), batch_size):
            kwargs["FL"] = devices[i : i + batch_size]
            kwargs["getters"] = "get_facts"
            self.event(
                f"retrieving facts data for devices {', '.join(kwargs['FL'])}",
                resource=instance,
            )
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )
            for worker, results in data.items():
                if results["failed"]:
                    log.error(
                        f"{worker} get_facts failed, errors: {'; '.join(results['errors'])}"
                    )
                    continue
                for host, host_data in results["result"].items():
                    if host_data["napalm_get"]["failed"]:
                        log.error(
                            f"{host} facts update failed: '{host_data['napalm_get']['exception']}'"
                        )
                        self.event(
                            f"{host} facts update failed",
                            resource=instance,
                            status="failed",
                            severity="WARNING",
                        )
                        continue
                    nb_device = nb.dcim.devices.get(name=host)
                    if not nb_device:
                        raise Exception(f"'{host}' does not exist in Netbox")
                    facts = host_data["napalm_get"]["result"]["get_facts"]
                    # update serial number
                    nb_device.serial = facts["serial_number"]
                    if not dry_run:
                        nb_device.save()
                    result[host] = {
                        (
                            "update_device_facts_dry_run"
                            if dry_run
                            else "update_device_facts"
                        ): {
                            "serial": facts["serial_number"],
                        }
                    }
                    self.event(f"{host} facts updated", resource=instance)
    else:
        raise UnsupportedServiceError(
            f"'{datasource}' datasource service not supported"
        )

    return ret

update_device_interfaces(instance: str = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: list = None, create: bool = True, batch_size: int = 10, **kwargs) -> dict

Update or create device interfaces in Netbox. Interface parameters updated:

• interface name
• interface description
• mtu
• mac address
• admin status
• speed

Parameters:

Name	Type	Description	Default
instance	str	The instance name to use.	None
dry_run	bool	If True, no changes will be made to Netbox.	False
datasource	str	The data source to use. Supported datasources: nornir - uses Nornir Service parse task to retrieve devices' data using NAPALM get_interfaces getter	'nornir'
timeout	int	The timeout for the job.	60
devices	list	List of devices to update.	None
create	bool	If True, new interfaces will be created if they do not exist.	True
batch_size	int	The number of devices to process in each batch.	10
**kwargs		Additional keyword arguments to pass to the datasource job.	{}

Returns:

Name	Type	Description
dict	dict	A dictionary containing the results of the update operation.

Raises:

Type	Description
Exception	If a device does not exist in Netbox.
UnsupportedServiceError	If the specified datasource is not supported.

Source code in norfab\workers\netbox_worker.py

def update_device_interfaces(
    self,
    instance: str = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: list = None,
    create: bool = True,
    batch_size: int = 10,
    **kwargs,
) -> dict:
    """
    Update or create device interfaces in Netbox. Interface parameters updated:

    - interface name
    - interface description
    - mtu
    - mac address
    - admin status
    - speed

    Args:
        instance (str, optional): The instance name to use.
        dry_run (bool, optional): If True, no changes will be made to Netbox.
        datasource (str, optional): The data source to use. Supported datasources:

            - **nornir** - uses Nornir Service parse task to retrieve devices' data
                using NAPALM get_interfaces getter

        timeout (int, optional): The timeout for the job.
        devices (list, optional): List of devices to update.
        create (bool, optional): If True, new interfaces will be created if they do not exist.
        batch_size (int, optional): The number of devices to process in each batch.
        **kwargs: Additional keyword arguments to pass to the datasource job.

    Returns:
        dict: A dictionary containing the results of the update operation.

    Raises:
        Exception: If a device does not exist in Netbox.
        UnsupportedServiceError: If the specified datasource is not supported.
    """
    result = {}
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:update_device_interfaces", result=result)
    nb = self._get_pynetbox(instance)

    if datasource == "nornir":
        # source hosts list from Nornir
        if kwargs:
            devices.extend(self.get_nornir_hosts(kwargs, timeout))
        # iterate over devices in batches
        for i in range(0, len(devices), batch_size):
            kwargs["FL"] = devices[i : i + batch_size]
            kwargs["getters"] = "get_interfaces"
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )
            for worker, results in data.items():
                if results["failed"]:
                    log.error(
                        f"{worker} get_interfaces failed, errors: {'; '.join(results['errors'])}"
                    )
                    continue
                for host, host_data in results["result"].items():
                    updated, created = {}, {}
                    result[host] = {
                        (
                            "update_device_interfaces_dry_run"
                            if dry_run
                            else "update_device_interfaces"
                        ): updated,
                        (
                            "created_device_interfaces_dry_run"
                            if dry_run
                            else "created_device_interfaces"
                        ): created,
                    }
                    interfaces = host_data["napalm_get"]["get_interfaces"]
                    nb_device = nb.dcim.devices.get(name=host)
                    if not nb_device:
                        raise Exception(f"'{host}' does not exist in Netbox")
                    nb_interfaces = nb.dcim.interfaces.filter(
                        device_id=nb_device.id
                    )
                    # update existing interfaces
                    for nb_interface in nb_interfaces:
                        if nb_interface.name not in interfaces:
                            continue
                        interface = interfaces.pop(nb_interface.name)
                        if dry_run is not True:
                            nb_interface.description = interface["description"]
                            if interface["mtu"] > 0:
                                nb_interface.mtu = interface["mtu"]
                            if interface["speed"] > 0:
                                nb_interface.speed = interface["speed"] * 1000
                            if interface["mac_address"].strip().lower() not in [
                                "none",
                                "",
                            ]:
                                mac_address = interface["mac_address"]
                                if self.nb_version[instance] >= (4, 2, 0):
                                    try:
                                        nb_mac_addr = nb.dcim.mac_addresses.get(
                                            mac_address=mac_address,
                                            interface_id=nb_interface.id,
                                        )
                                    except Exception as e:
                                        msg = f"{worker} failed to get {mac_address} mac-address from netbox: {e}"
                                        self.event(msg, severity="WARNING")
                                        ret.messages.append(msg)
                                        log.error(msg)
                                    else:
                                        if not nb_mac_addr:
                                            nb_mac_addr = nb.dcim.mac_addresses.create(
                                                mac_address=mac_address,
                                                assigned_object_type="dcim.interface",
                                                assigned_object_id=nb_interface.id,
                                            )
                                        elif not nb_mac_addr.assigned_object_id:
                                            nb_mac_addr.assigned_object_type = (
                                                "dcim.interface"
                                            )
                                            nb_mac_addr.assigned_object_id = (
                                                nb_interface.id
                                            )
                                        if not nb_interface.primary_mac_address:
                                            nb_interface.primary_mac_address = (
                                                nb_mac_addr.id
                                            )
                                            self.event(
                                                f"{host} {mac_address} MAC associating with interface {nb_interface.name}",
                                                resource=instance,
                                            )
                                    nb_interface.mac_address = mac_address
                            nb_interface.enabled = interface["is_enabled"]
                            nb_interface.save()
                        updated[nb_interface.name] = interface
                        self.event(
                            f"{host} updated interface {nb_interface.name}",
                            resource=instance,
                        )
                    # create new interfaces
                    if create is not True:
                        continue
                    for interface_name, interface in interfaces.items():
                        interface["type"] = "other"
                        if dry_run is not True:
                            nb_interface = nb.dcim.interfaces.create(
                                name=interface_name,
                                device={"name": nb_device.name},
                                type=interface["type"],
                            )
                            nb_interface.description = interface["description"]
                            if interface["mtu"] > 0:
                                nb_interface.mtu = interface["mtu"]
                            if interface["speed"] > 0:
                                nb_interface.speed = interface["speed"] * 1000
                            # create MAC address
                            if interface["mac_address"].strip().lower() not in [
                                "none",
                                "",
                            ]:
                                mac_address = interface["mac_address"]
                                if self.nb_version[instance] >= (4, 2, 0):
                                    try:
                                        nb_mac_addr = nb.dcim.mac_addresses.create(
                                            mac_address=mac_address,
                                            assigned_object_type="dcim.interface",
                                            assigned_object_id=nb_interface.id,
                                        )
                                    except Exception as e:
                                        msg = f"{host} {mac_address} MAC failed to create, interface {nb_interface.name}: {e}"
                                        self.event(msg, severity="WARNING")
                                        ret.messages.append(msg)
                                        log.error(msg)
                                    else:
                                        nb_interface.primary_mac_address = (
                                            nb_mac_addr.id
                                        )
                                else:
                                    nb_interface.mac_address = mac_address
                                self.event(
                                    f"{host} {mac_address} MAC created, associating with interface {nb_interface.name}",
                                    resource=instance,
                                )
                            nb_interface.enabled = interface["is_enabled"]
                            nb_interface.save()
                        created[interface_name] = interface
                        self.event(
                            f"{host} created interface {interface_name}",
                            resource=instance,
                        )
    else:
        raise UnsupportedServiceError(
            f"'{datasource}' datasource service not supported"
        )

    return ret

update_device_ip(instance: str = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: list = None, create: bool = True, batch_size: int = 10, **kwargs) -> dict

Update the IP addresses of devices in Netbox.

Parameters:

Name	Type	Description	Default
instance	str	The instance name to use.	None
dry_run	bool	If True, no changes will be made.	False
datasource	str	The data source to use. Supported datasources: nornir - uses Nornir Service parse task to retrieve devices' data using NAPALM get_interfaces_ip getter	'nornir'
timeout	int	The timeout for the operation.	60
devices	list	The list of devices to update.	None
create	bool	If True, new IP addresses will be created if they do not exist.	True
batch_size	int	The number of devices to process in each batch.	10
**kwargs		Additional keyword arguments.	{}

Returns:

Name	Type	Description
dict	dict	A dictionary containing the results of the update operation.

Raises:

Type	Description
Exception	If a device does not exist in Netbox.
UnsupportedServiceError	If the specified datasource is not supported.

Source code in norfab\workers\netbox_worker.py

def update_device_ip(
    self,
    instance: str = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: list = None,
    create: bool = True,
    batch_size: int = 10,
    **kwargs,
) -> dict:
    """
    Update the IP addresses of devices in Netbox.

    Args:
        instance (str, optional): The instance name to use.
        dry_run (bool, optional): If True, no changes will be made.
        datasource (str, optional): The data source to use. Supported datasources:

            - **nornir** - uses Nornir Service parse task to retrieve devices' data
                using NAPALM get_interfaces_ip getter

        timeout (int, optional): The timeout for the operation.
        devices (list, optional): The list of devices to update.
        create (bool, optional): If True, new IP addresses will be created if they do not exist.
        batch_size (int, optional): The number of devices to process in each batch.
        **kwargs: Additional keyword arguments.

    Returns:
        dict: A dictionary containing the results of the update operation.

    Raises:
        Exception: If a device does not exist in Netbox.
        UnsupportedServiceError: If the specified datasource is not supported.
    """
    result = {}
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:update_device_ip", result=result)
    nb = self._get_pynetbox(instance)

    if datasource == "nornir":
        # source hosts list from Nornir
        if kwargs:
            devices.extend(self.get_nornir_hosts(kwargs, timeout))
        # iterate over devices in batches
        for i in range(0, len(devices), batch_size):
            kwargs["FL"] = devices[i : i + batch_size]
            kwargs["getters"] = "get_interfaces_ip"
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )
            for worker, results in data.items():
                if results["failed"]:
                    log.error(
                        f"{worker} get_interfaces_ip failed, errors: {'; '.join(results['errors'])}"
                    )
                    continue
                for host, host_data in results["result"].items():
                    updated, created = {}, {}
                    result[host] = {
                        "updated_ip_dry_run" if dry_run else "updated_ip": updated,
                        "created_ip_dry_run" if dry_run else "created_ip": created,
                    }
                    interfaces = host_data["napalm_get"]["get_interfaces_ip"]
                    nb_device = nb.dcim.devices.get(name=host)
                    if not nb_device:
                        raise Exception(f"'{host}' does not exist in Netbox")
                    nb_interfaces = nb.dcim.interfaces.filter(
                        device_id=nb_device.id
                    )
                    # update interface IP addresses
                    for nb_interface in nb_interfaces:
                        if nb_interface.name not in interfaces:
                            continue
                        interface = interfaces.pop(nb_interface.name)
                        # merge v6 into v4 addresses to save code repetition
                        ips = {
                            **interface.get("ipv4", {}),
                            **interface.get("ipv6", {}),
                        }
                        # update/create IP addresses
                        for ip, ip_data in ips.items():
                            prefix_length = ip_data["prefix_length"]
                            # get IP address info from Netbox
                            nb_ip = nb.ipam.ip_addresses.filter(
                                address=f"{ip}/{prefix_length}"
                            )
                            if len(nb_ip) > 1:
                                log.warning(
                                    f"{host} got multiple {ip}/{prefix_length} IP addresses from Netbox, "
                                    f"NorFab Netbox Service only supports handling of non-duplicate IPs."
                                )
                                continue
                            # decide what to do
                            if not nb_ip and create is False:
                                continue
                            elif not nb_ip and create is True:
                                if dry_run is not True:
                                    try:
                                        nb_ip = nb.ipam.ip_addresses.create(
                                            address=f"{ip}/{prefix_length}"
                                        )
                                    except Exception as e:
                                        msg = f"{host} failed to create {ip}/{prefix_length}, error: {e}"
                                        log.error(msg)
                                        self.event(msg, resource=instance)
                                        continue
                                    nb_ip.assigned_object_type = "dcim.interface"
                                    nb_ip.assigned_object_id = nb_interface.id
                                    nb_ip.status = "active"
                                    nb_ip.save()
                                created[f"{ip}/{prefix_length}"] = nb_interface.name
                                self.event(
                                    f"{host} created IP address {ip}/{prefix_length} for {nb_interface.name} interface",
                                    resource=instance,
                                )
                            elif nb_ip:
                                nb_ip = list(nb_ip)[0]
                                if dry_run is not True:
                                    nb_ip.assigned_object_type = "dcim.interface"
                                    nb_ip.assigned_object_id = nb_interface.id
                                    nb_ip.status = "active"
                                    nb_ip.save()
                                updated[nb_ip.address] = nb_interface.name
                                self.event(
                                    f"{host} updated IP address {ip}/{prefix_length} for {nb_interface.name} interface",
                                    resource=instance,
                                )

    else:
        raise UnsupportedServiceError(
            f"'{datasource}' datasource service not supported"
        )

    return ret

get_next_ip(subnet: str, description: str = None, device: str = None, interface: str = None, vrf: str = None, tags: list = None, dns_name: str = None, tenant: str = None, comments: str = None, instance: str = None, dry_run: bool = False) -> dict

Allocate the next available IP address from a given subnet.

Parameters:

Name	Type	Description	Default
subnet	str	The subnet from which to allocate the IP address.	required
description	str	A description for the allocated IP address.	None
device	str	The device associated with the IP address.	None
interface	str	The interface associated with the IP address.	None
vrf	str	The VRF (Virtual Routing and Forwarding) instance.	None
tags	list	A list of tags to associate with the IP address.	None
dns_name	str	The DNS name for the IP address.	None
tenant	str	The tenant associated with the IP address.	None
comments	str	Additional comments for the IP address.	None
instance	str	The NetBox instance to use.	None
dry_run	bool	If True, do not actually allocate the IP address. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	A dictionary containing the result of the IP allocation.

Source code in norfab\workers\netbox_worker.py

def get_next_ip(
    self,
    subnet: str,
    description: str = None,
    device: str = None,
    interface: str = None,
    vrf: str = None,
    tags: list = None,
    dns_name: str = None,
    tenant: str = None,
    comments: str = None,
    instance: str = None,
    dry_run: bool = False,
) -> dict:
    """
    Allocate the next available IP address from a given subnet.

    Args:
        subnet (str): The subnet from which to allocate the IP address.
        description (str, optional): A description for the allocated IP address.
        device (str, optional): The device associated with the IP address.
        interface (str, optional): The interface associated with the IP address.
        vrf (str, optional): The VRF (Virtual Routing and Forwarding) instance.
        tags (list, optional): A list of tags to associate with the IP address.
        dns_name (str, optional): The DNS name for the IP address.
        tenant (str, optional): The tenant associated with the IP address.
        comments (str, optional): Additional comments for the IP address.
        instance (str, optional): The NetBox instance to use.
        dry_run (bool, optional): If True, do not actually allocate the IP address. Defaults to False.

    Returns:
        dict: A dictionary containing the result of the IP allocation.
    """
    nb = self._get_pynetbox(instance)
    nb_prefix = nb.ipam.prefixes.get(prefix=subnet, vrf=vrf)
    nb_ip = nb_prefix.available_ips.create()
    if description is not None:
        nb_ip.description = description
    nb_ip.save()

    return Result(result=str(nb_ip))

get_containerlab_inventory(lab_name: str = None, tenant: str = None, filters: list = None, devices: list = None, instance: str = None, image: str = None, ipv4_subnet: str = '172.100.100.0/24', ports: tuple = (12000, 15000), ports_map: dict = None, progress: bool = False, cache: Union[bool, str] = False) -> dict

Retrieve and construct Containerlab inventory from NetBox data.

Containerlab node details must be defined under device configuration context norfab.containerlab path, for example:

{
    "norfab": {
        "containerlab": {
            "kind": "ceos",
            "image": "ceos:latest",
            "mgmt-ipv4": "172.100.100.10/24",
            "ports": [
                {10000: 22},
                {10001: 830}
            ],

            ... any other node parameters ...

            "interfaces_rename": [
                {
                    "find": "eth",
                    "replace": "Eth",
                    "use_regex": false
                }
            ]
        }
    }
}

For complete list of parameters refer to Containerlab nodes definition.

Special handling given to these parameters:

• lab_name - if not provided uses tenant argument value as a lab name
• kind - uses device platform field value by default
• image - uses image value if provided, otherwise uses {kind}:latest
• interfaces_rename - a list of one or more interface renaming instructions, each item must have find and replace defined, optional use_regex flag specifies whether to use regex based pattern substitution.

To retrieve topology data from Netbox at least one of these arguments must be provided to identify a set of devices to include into Containerlab topology:

• tenant - topology constructed using all devices and links that belong to this tenant
• devices - creates topology only using devices in the lists
• filters - list of device filters to retrieve from Netbox and add to topology

If multiple of above arguments provided, resulting lab topology is a sum of all devices matched.

Parameters:

Name	Type	Description	Default
lab_name	(str, Mandatory)	Name of containerlab to construct inventory for.	None
tenant	str	Construct topology using given tenant's devices	None
filters	list	List of filters to apply when retrieving devices from NetBox.	None
devices	list	List of specific devices to retrieve from NetBox.	None
instance	str	NetBox instance to use.	None
image	str	Default containerlab image to use,	None
ipv4_subnet	(str, Optional)	Management subnet to use to IP number nodes starting with 2nd IP in the subnet, in assumption that 1st IP is a default gateway.	'172.100.100.0/24'
ports	(tuple, Optional)	Ports range to use for nodes.	(12000, 15000)
ports_map	(dict, Optional)	dictionary keyed by node name with list of ports maps to use,	None
progress	bool	If True, emits progress events.	False
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.	False

Returns:

Name	Type	Description
dict	dict	Containerlab inventory dictionary containing lab topology data

Source code in norfab\workers\netbox_worker.py

def get_containerlab_inventory(
    self,
    lab_name: str = None,
    tenant: str = None,
    filters: list = None,
    devices: list = None,
    instance: str = None,
    image: str = None,
    ipv4_subnet: str = "172.100.100.0/24",
    ports: tuple = (12000, 15000),
    ports_map: dict = None,
    progress: bool = False,
    cache: Union[bool, str] = False,
) -> dict:
    """
    Retrieve and construct Containerlab inventory from NetBox data.

    Containerlab node details must be defined under device configuration
    context `norfab.containerlab` path, for example:

    ```
    {
        "norfab": {
            "containerlab": {
                "kind": "ceos",
                "image": "ceos:latest",
                "mgmt-ipv4": "172.100.100.10/24",
                "ports": [
                    {10000: 22},
                    {10001: 830}
                ],

                ... any other node parameters ...

                "interfaces_rename": [
                    {
                        "find": "eth",
                        "replace": "Eth",
                        "use_regex": false
                    }
                ]
            }
        }
    }
    ```

    For complete list of parameters refer to
    [Containerlab nodes definition](https://containerlab.dev/manual/nodes/).

    Special handling given to these parameters:

    - `lab_name` - if not provided uses `tenant` argument value as a lab name
    - `kind` - uses device platform field value by default
    - `image` - uses `image` value if provided, otherwise uses `{kind}:latest`
    - `interfaces_rename` - a list of one or more interface renaming instructions,
        each item must have `find` and `replace` defined, optional `use_regex`
        flag specifies whether to use regex based pattern substitution.

    To retrieve topology data from Netbox at least one of these arguments must be provided
    to identify a set of devices to include into Containerlab topology:

    - `tenant` - topology constructed using all devices and links that belong to this tenant
    - `devices` - creates topology only using devices in the lists
    - `filters` - list of device filters to retrieve from Netbox and add to topology

    If multiple of above arguments provided, resulting lab topology is a sum of all
    devices matched.

    Args:
        lab_name (str, Mandatory): Name of containerlab to construct inventory for.
        tenant (str, optional): Construct topology using given tenant's devices
        filters (list, optional): List of filters to apply when retrieving devices from NetBox.
        devices (list, optional): List of specific devices to retrieve from NetBox.
        instance (str, optional): NetBox instance to use.
        image (str, optional): Default containerlab image to use,
        ipv4_subnet (str, Optional): Management subnet to use to IP number nodes
            starting with 2nd IP in the subnet, in assumption that 1st IP is a default gateway.
        ports (tuple, Optional): Ports range to use for nodes.
        ports_map (dict, Optional): dictionary keyed by node name with list of ports maps to use,
        progress (bool, optional): If True, emits progress events.
        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: Containerlab inventory dictionary containing lab topology data
    """
    nodes, links = {}, []
    ports_map = ports_map or {}
    endpts_done = []  # to deduplicate links
    instance = instance or self.default_instance
    # handle lab name and tenant name with filters
    if lab_name is None and tenant:
        lab_name = tenant
    if tenant:
        if filters:
            for filter in filters:
                filter["tenant"] = tenant
        else:
            filters = [{"tenant": tenant}]
    # construct inventory
    inventory = {
        "name": lab_name,
        "topology": {"nodes": nodes, "links": links},
        "mgmt": {"ipv4-subnet": ipv4_subnet, "network": f"br-{lab_name}"},
    }
    ret = Result(task=f"{self.name}:get_containerlab_inventory", result=inventory)
    mgmt_net = ipaddress.ip_network(ipv4_subnet)
    available_ips = list(mgmt_net.hosts())[1:]

    # run checks
    if not available_ips:
        raise ValueError(f"Need IPs to allocate, but '{ipv4_subnet}' given")
    if ports:
        available_ports = list(range(ports[0], ports[1]))
    else:
        raise ValueError(f"Need ports to allocate, but '{ports}' given")

    # check Netbox status
    netbox_status = self.get_netbox_status(instance=instance)
    if netbox_status.result[instance]["status"] is False:
        ret.failed = True
        ret.messages = [f"Netbox status is no good: {netbox_status}"]
        return ret

    # retrieve devices data
    log.debug(
        f"Fetching devices from {instance} Netbox instance, devices '{devices}', filters '{filters}'"
    )
    if progress:
        self.event("Fetching devices data from Netbox")
    nb_devices = self.get_devices(
        filters=filters, devices=devices, instance=instance, cache=cache
    )

    # form Containerlab nodes inventory
    for device_name, device in nb_devices.result.items():
        node = device["config_context"].get("norfab", {}).get("containerlab", {})
        # populate node parameters
        if not node.get("kind"):
            if device["platform"]:
                node["kind"] = device["platform"]["name"]
            else:
                msg = (
                    f"{device_name} - has no 'kind' of 'platform' defined, skipping"
                )
                log.warning(msg)
                if progress:
                    self.event(msg, severity="WARNING")
                continue
        if not node.get("image"):
            if image:
                node["image"] = image
            else:
                node["image"] = f"{node['kind']}:latest"
        if not node.get("mgmt-ipv4"):
            if available_ips:
                node["mgmt-ipv4"] = f"{available_ips.pop(0)}"
            else:
                raise RuntimeError("Run out of IP addresses to allocate")
        if not node.get("ports"):
            node["ports"] = []
            # use ports map
            if ports_map.get(device_name):
                node["ports"] = ports_map[device_name]
            # allocate next-available ports
            else:
                for port in [
                    "22/tcp",
                    "23/tcp",
                    "80/tcp",
                    "161/udp",
                    "443/tcp",
                    "830/tcp",
                    "8080/tcp",
                ]:
                    if available_ports:
                        node["ports"].append(f"{available_ports.pop(0)}:{port}")
                    else:
                        raise RuntimeError(
                            "Run out of TCP / UDP ports to allocate."
                        )

        # save node content
        nodes[device_name] = node
        if progress:
            self.event(f"Node added {device_name}")

    # return if no nodes found for provided parameters
    if not nodes:
        msg = f"{self.name} - no devices found in Netbox"
        log.error(msg)
        ret.failed = True
        ret.messages = [
            f"{self.name} - no devices found in Netbox, "
            f"devices - '{devices}', filters - '{filters}'"
        ]
        ret.errors = [msg]
        return ret

    if progress:
        self.event("Fetching connections data from Netbox")

    # query interface connections data from netbox
    nb_connections = self.get_connections(
        devices=list(nodes), instance=instance, cache=cache
    )
    # save connections data to links inventory
    while nb_connections.result:
        device, device_connections = nb_connections.result.popitem()
        for interface, connection in device_connections.items():
            # skip non ethernet links
            if connection.get("termination_type") != "interface":
                continue
            # skip orphaned links
            if not connection.get("remote_interface"):
                continue
            # skip connections to devices that are not part of lab
            if connection["remote_device"] not in nodes:
                continue
            endpoints = []
            link = {
                "type": "veth",
                "endpoints": endpoints,
            }
            # add A node
            endpoints.append(
                {
                    "node": device,
                    "interface": interface,
                }
            )
            # add B node
            endpoints.append({"node": connection["remote_device"]})
            if connection.get("breakout") is True:
                endpoints[-1]["interface"] = connection["remote_interface"][0]
            else:
                endpoints[-1]["interface"] = connection["remote_interface"]
            # save the link
            a_end = (
                endpoints[0]["node"],
                endpoints[0]["interface"],
            )
            b_end = (
                endpoints[1]["node"],
                endpoints[1]["interface"],
            )
            if a_end not in endpts_done and b_end not in endpts_done:
                endpts_done.append(a_end)
                endpts_done.append(b_end)
                links.append(link)
                if progress:
                    self.event(
                        f"Link added {endpoints[0]['node']}:{endpoints[0]['interface']}"
                        f" - {endpoints[1]['node']}:{endpoints[1]['interface']}"
                    )

    # query circuits connections data from netbox
    nb_circuits = self.get_circuits(
        devices=list(nodes), instance=instance, cache=cache
    )
    # save circuits data to hosts' inventory
    while nb_circuits.result:
        device, device_circuits = nb_circuits.result.popitem()
        for cid, circuit in device_circuits.items():
            # skip circuits not connected to devices
            if not circuit.get("remote_interface"):
                continue
            # skip circuits to devices that are not part of lab
            if circuit["remote_device"] not in nodes:
                continue
            endpoints = []
            link = {
                "type": "veth",
                "endpoints": endpoints,
            }
            # add A node
            endpoints.append(
                {
                    "node": device,
                    "interface": circuit["interface"],
                }
            )
            # add B node
            endpoints.append(
                {
                    "node": circuit["remote_device"],
                    "interface": circuit["remote_interface"],
                }
            )
            # save the link
            a_end = (
                endpoints[0]["node"],
                endpoints[0]["interface"],
            )
            b_end = (
                endpoints[1]["node"],
                endpoints[1]["interface"],
            )
            if a_end not in endpts_done and b_end not in endpts_done:
                endpts_done.append(a_end)
                endpts_done.append(b_end)
                links.append(link)
                if progress:
                    self.event(
                        f"Link added {endpoints[0]['node']}:{endpoints[0]['interface']}"
                        f" - {endpoints[1]['node']}:{endpoints[1]['interface']}"
                    )

    # rename links' interfaces
    for node_name, node_data in nodes.items():
        interfaces_rename = node_data.pop("interfaces_rename", [])
        if interfaces_rename and progress:
            self.event(f"Renaming {node_name} interfaces")
        for item in interfaces_rename:
            if not item.get("find") or not item.get("replace"):
                log.error(
                    f"{self.name} - interface rename need to have"
                    f" 'find' and 'replace' defined, skipping: {item}"
                )
                continue
            pattern = item["find"]
            replace = item["replace"]
            use_regex = item.get("use_regex", False)
            # go over links one by one and rename interfaces
            for link in links:
                for endpoint in link["endpoints"]:
                    if endpoint["node"] != node_name:
                        continue
                    if use_regex:
                        renamed = re.sub(
                            pattern,
                            replace,
                            endpoint["interface"],
                        )
                    else:
                        renamed = endpoint["interface"].replace(pattern, replace)
                    if endpoint["interface"] != renamed:
                        msg = f"{node_name} interface {endpoint['interface']} renamed to {renamed}"
                        log.debug(msg)
                        if progress:
                            self.event(msg)
                        endpoint["interface"] = renamed

    return ret