Netbox Worker

NetboxAllocationError

Bases: Exception

Raised when there is an error in allocating resource in Netbox

UnsupportedNetboxVersion

Bases: Exception

Raised when there is an error in allocating resource in Netbox

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_v4	tuple	Minimum supported Netbox v4 version (4.4.0+).

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
    self.branch_create_timeout = self.netbox_inventory.get(
        "branch_create_timeout", 120
    )

    # 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() -> Result

NorFab Task to return running inventory for NetBox worker.

Returns:

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

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_inventory(self) -> Result:
    """
    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: Any) -> Result

Retrieves the version information of Netbox instances.

Returns:

Name	Type	Description
dict	Result	A dictionary containing the version information of the Netbox

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_version(self, **kwargs: Any) -> Result:
    """
    Retrieves the version information of Netbox instances.

    Returns:
        dict: A dictionary containing the version information of the Netbox
    """
    libs = {
        "norfab": "",
        "pynetbox": "",
        "requests": "",
        "python": sys.version.split(" ")[0],
        "platform": sys.platform,
        "diskcache": "",
        "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: Union[None, str] = None) -> Result

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	Result	A dictionary containing the status of the requested NetBox instance(s).

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_netbox_status(self, instance: Union[None, str] = None) -> Result:
    """
    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(job: Job) -> Result

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.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required

Returns:

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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_compatibility(self, job: Job) -> Result:
    """
    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.

    Args:
        job: NorFab Job object containing relevant metadata

    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(job=job)
    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(".")]
                )
            # check Netbox 4.4+ compatibility
            if self.nb_version[instance] >= self.compatible_ge_v4:
                ret.result[instance] = True
            else:
                ret.result[instance] = False
                log.error(
                    f"{self.name} - {instance} Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )

    return ret

has_plugin(plugin_name: str, instance: str, strict: bool = False) -> bool

Check if a specified plugin is installed in a given NetBox instance.

Parameters:

Name	Type	Description	Default
plugin_name	str	The name of the plugin to check for.	required
instance	str	The identifier or address of the NetBox instance.	required
strict	bool	If True, raises a RuntimeError when the plugin is not found.	False

Returns:

Name	Type	Description
bool	bool	True if the plugin is installed, False otherwise.

Source code in norfab\workers\netbox_worker.py

def has_plugin(self, plugin_name: str, instance: str, strict: bool = False) -> bool:
    """
    Check if a specified plugin is installed in a given NetBox instance.

    Args:
        plugin_name (str): The name of the plugin to check for.
        instance (str): The identifier or address of the NetBox instance.
        strict (bool, optional): If True, raises a RuntimeError when the plugin is not found.

    Returns:
        bool: True if the plugin is installed, False otherwise.
    """
    nb_status = self._query_netbox_status(instance)

    if plugin_name in nb_status["plugins"]:
        return True
    elif strict is True:
        raise RuntimeError(
            f"'{instance}' Netbox instance has no '{plugin_name}' plugin installed"
        )

    return False

cache_list(keys: str = '*', details: bool = False) -> Result

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	Result	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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def cache_list(self, keys: str = "*", details: bool = False) -> Result:
    """
    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(job: Job, key: str = None, keys: str = None) -> Result

Clears specified keys from the cache.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
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	Result	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

@Task(
    fastapi={"methods": ["DELETE"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def cache_clear(self, job: Job, key: str = None, keys: str = None) -> Result:
    """
    Clears specified keys from the cache.

    Args:
        job: NorFab Job object containing relevant metadata
        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:  # noqa
        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(job: Job, key: str = None, keys: str = None, raise_missing: bool = False) -> Result

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

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
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	Result	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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def cache_get(
    self, job: Job, key: str = None, keys: str = None, raise_missing: bool = False
) -> Result:
    """
    Retrieve values from the cache based on a specific key or a pattern of keys.

    Args:
        job: NorFab Job object containing relevant metadata
        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(job: Job, instance: Union[None, str] = None, dry_run: bool = False, obj: Union[str, dict] = None, filters: Union[None, dict, str] = None, fields: Union[None, list] = None, queries: Union[None, dict] = None, query_string: str = None) -> Result

Function to query Netbox v3 or Netbox v4 GraphQL API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	Union[None, str]	Netbox instance name	None
dry_run	bool	only return query content, do not run it	False
obj	Union[str, dict]	Object to query	None
filters	Union[None, dict, str]	Filters to apply to the query	None
fields	Union[None, list]	Fields to retrieve in the query	None
queries	Union[None, dict]	Dictionary of queries to execute	None
query_string	str	Raw query string to execute	None

Returns:

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def graphql(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    obj: Union[str, dict] = None,
    filters: Union[None, dict, str] = None,
    fields: Union[None, list] = None,
    queries: Union[None, dict] = None,
    query_string: str = None,
) -> Result:
    """
    Function to query Netbox v3 or Netbox v4 GraphQL API.

    Args:
        job: NorFab Job object containing relevant metadata
        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)
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:graphql", resources=[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] >= (4, 4, 0):
                queries_list.append(_form_query_v4(**query_data))
            else:
                raise UnsupportedNetboxVersion(
                    f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )
        queries_strings = "    ".join(queries_list)
        query = f"query {{{queries_strings}}}"
    elif obj and filters and fields:
        if self.nb_version[instance] >= (4, 4, 0):
            query = _form_query_v4(obj, filters, fields)
        else:
            raise UnsupportedNetboxVersion(
                f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                f"minimum required version is {self.compatible_ge_v4}"
            )
        query = f"query {{{query}}}"
    elif query_string:
        query = query_string
    else:
        raise RuntimeError(
            f"{self.name} - graphql method expects queries 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:
        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(job: Job, instance: Union[None, str] = None, method: str = 'get', api: str = '', **kwargs: Any) -> Result

Sends a request to the Netbox REST API.

Parameters:

Name	Type	Description	Default
instance	str	The Netbox 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	Any	Additional arguments to pass to the request (e.g., params, data, json).	{}

Returns:

Type	Description
Result	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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def rest(
    self,
    job: Job,
    instance: Union[None, str] = None,
    method: str = "get",
    api: str = "",
    **kwargs: Any,
) -> Result:
    """
    Sends a request to the Netbox REST API.

    Args:
        instance (str, optional): The Netbox 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.
    """
    ret = Result(task=f"{self.name}:rest", result={})
    nb_params = self._get_instance_params(instance)

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

    response.raise_for_status()
    try:
        ret.result = response.json()
    except Exception as e:
        log.debug(f"Failed to decode json, error: {e}")
        ret.result = response.text if response.text else response.status_code

    return ret

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

Retrieves device data from Netbox using the GraphQL API.

Parameters:

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

Returns:

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

Raises:

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

Source code in norfab\workers\netbox_worker.py

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

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

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

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

    Raises:
        Exception: If the GraphQL query fails or if there are errors in the query result.
    """
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:get_devices", result={}, resources=[instance])
    cache = self.cache_use if cache is None else cache
    filters = filters or []
    devices = devices or []
    queries = {}  # devices queries
    device_fields = [
        "name",
        "last_updated",
        "custom_field_data",
        "tags {name}",
        "device_type {model}",
        "role {name}",
        "config_context",
        "tenant {name}",
        "platform {name}",
        "serial",
        "asset_tag",
        "site {name slug tags{name} }",
        "location {name}",
        "rack {name}",
        "status",
        "primary_ip4 {address}",
        "primary_ip6 {address}",
        "airflow",
        "position",
        "id",
    ]

    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] >= (4, 4, 0):
                dlist = '["{dl}"]'.format(dl='", "'.join(devices))
                filters_dict = {"name": f"{{in_list: {dlist}}}"}
            else:
                raise UnsupportedNetboxVersion(
                    f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )
            last_updated_query["devices_by_devices_list"] = {
                "obj": "device_list",
                "filters": filters_dict,
                "fields": ["name", "last_updated"],
            }
        last_updated = self.graphql(
            job=job,
            queries=last_updated_query,
            instance=instance,
            dry_run=dry_run,
        )
        last_updated.raise_for_status(f"{self.name} - get devices query failed")

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

        # try to retrieve device data from cache
        self.cache.expire()  # remove expired items from cache
        for devices_list in last_updated.result.values():
            for device in devices_list:
                device_cache_key = f"get_devices::{device['name']}"
                # check if cache is up to date and use it if so
                if device_cache_key in self.cache and (
                    self.cache[device_cache_key].get("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] >= (4, 4, 0):
                dlist = '["{dl}"]'.format(dl='", "'.join(devices))
                filters_dict = {"name": f"{{in_list: {dlist}}}"}
            else:
                raise UnsupportedNetboxVersion(
                    f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )
            queries["devices_by_devices_list"] = {
                "obj": "device_list",
                "filters": filters_dict,
                "fields": device_fields,
            }

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

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

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

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

    return ret

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

Retrieve device interfaces from Netbox using GraphQL API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	str	Netbox instance name.	None
devices	list	List of devices to retrieve interfaces for.	None
interface_regex	str	Regex pattern to match interfaces by name, case insensitive.	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	Result	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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_interfaces(
    self,
    job: Job,
    instance: Union[None, str] = None,
    devices: Union[None, list] = None,
    interface_regex: Union[None, str] = None,
    ip_addresses: bool = False,
    inventory_items: bool = False,
    dry_run: bool = False,
    cache: Union[bool, str] = None,
) -> Result:
    """
    Retrieve device interfaces from Netbox using GraphQL API.

    Args:
        job: NorFab Job object containing relevant metadata
        instance (str, optional): Netbox instance name.
        devices (list, optional): List of devices to retrieve interfaces for.
        interface_regex (str, optional): Regex pattern to match interfaces by name, case insensitive.
        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.
    """
    instance = instance or self.default_instance
    devices = devices or []
    ret = Result(
        task=f"{self.name}:get_interfaces",
        result={d: {} for d in devices},
        resources=[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}",
        "label",
        "mark_connected",
    ]
    intf_fields.append("mac_addresses {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
    dlist = str(devices).replace("'", '"')  # swap quotes
    if self.nb_version[instance] >= (4, 4, 0):
        # add interface name regex filter
        if interface_regex:
            filters = (
                "{device: {name: {in_list: "
                + dlist
                + "}}"
                + ", name: {i_regex: "
                + f'"{interface_regex}"'
                + "}}"
            )
        else:
            filters = "{device: {name: {in_list: " + dlist + "}}}"
    else:
        raise UnsupportedNetboxVersion(
            f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
            f"minimum required version is {self.compatible_ge_v4}"
        )

    queries = {
        "interfaces": {
            "obj": "interface_list",
            "filters": filters,
            "fields": intf_fields,
        }
    }

    # add query to retrieve inventory items
    if inventory_items:
        if self.nb_version[instance] >= (4, 4, 0):
            dlist = str(devices).replace("'", '"')  # swap quotes
            inv_filters = (
                "{device: {name: {in_list: "
                + dlist
                + '}}, component_type: {app_label: {exact: "dcim"}}}'
            )
        else:
            raise UnsupportedNetboxVersion(
                f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                f"minimum required version is {self.compatible_ge_v4}"
            )
        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(
        job=job, 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()
        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(job: Job, devices: list[str], instance: Union[None, str] = None, dry_run: bool = False, cables: bool = False, cache: Union[bool, str] = None, include_virtual: bool = True, interface_regex: Union[None, str] = None) -> Result

Retrieve interface connection details for specified devices from Netbox.

This task retrieves these connections:

• Physical interfaces connections
• Child/virtual interfaces connections using parent interface connections details
• Lag interfaces connections using member ports connections details
• Lag child interfaces connections using member ports connections details
• Console port and console server ports connections
• Connections to provider networks for physical, child/virtual and lag interfaces

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
devices	list	List of device names to retrieve connections for.	required
instance	str	Netbox 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
include_virtual	bool	if True include connections for virtual and LAG interfaces	True
interface_regex	str	Regex pattern to match interfaces, console ports and console server ports by name, case insensitive.	None

Returns:

Name	Type	Description
dict	Result	A dictionary containing connection details for each device: { "netbox-worker-1.2": { "r1": { "Console": { "breakout": false, "remote_device": "termserv1", "remote_device_status": "active", "remote_interface": "ConsoleServerPort1", "remote_termination_type": "consoleserverport", "termination_type": "consoleport" }, "eth1": { "breakout": false, "remote_device": "r2", "remote_device_status": "active", "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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_connections(
    self,
    job: Job,
    devices: list[str],
    instance: Union[None, str] = None,
    dry_run: bool = False,
    cables: bool = False,
    cache: Union[bool, str] = None,
    include_virtual: bool = True,
    interface_regex: Union[None, str] = None,
) -> Result:
    """
    Retrieve interface connection details for specified devices from Netbox.

    This task retrieves these connections:

    - Physical interfaces connections
    - Child/virtual interfaces connections using parent interface connections details
    - Lag interfaces connections using member ports connections details
    - Lag child interfaces connections using member ports connections details
    - Console port and console server ports connections
    - Connections to provider networks for physical, child/virtual and lag interfaces

    Args:
        job: NorFab Job object containing relevant metadata
        devices (list): List of device names to retrieve connections for.
        instance (str, optional): Netbox 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
        include_virtual (bool, optional): if True include connections for virtual and LAG interfaces
        interface_regex (str, optional): Regex pattern to match interfaces, console ports and
            console server ports by name, case insensitive.

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

            ```
            {
                "netbox-worker-1.2": {
                    "r1": {
                        "Console": {
                            "breakout": false,
                            "remote_device": "termserv1",
                            "remote_device_status": "active",
                            "remote_interface": "ConsoleServerPort1",
                            "remote_termination_type": "consoleserverport",
                            "termination_type": "consoleport"
                        },
                        "eth1": {
                            "breakout": false,
                            "remote_device": "r2",
                            "remote_device_status": "active",
                            "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.
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:get_connections",
        result={d: {} for d in devices},
        resources=[instance],
    )

    # form lists of fields to request from netbox
    cable_fields = """
        cable {
            type
            status
            tenant {name}
            label
            tags {name}
            length
            length_unit
            custom_fields
        }
    """
    interfaces_fields = [
        "name",
        "type",
        "device {name, status}",
        """
        member_interfaces {
          name
          connected_endpoints {
            __typename
            ... on ProviderNetworkType {name}
            ... on InterfaceType {name, device {name, status}, child_interfaces {name}, lag {name child_interfaces {name}}}
          }
        }
        """,
        """
        parent {
          name
          type
          member_interfaces {
            name
            connected_endpoints {
              __typename
              ... on ProviderNetworkType {name}
              ... on InterfaceType {name, device {name, status}, child_interfaces {name}, lag {name child_interfaces {name}}}
            }
          }
          connected_endpoints {
            __typename
            ... on ProviderNetworkType {name}
            ... on InterfaceType {name, device {name, status}, child_interfaces {name}, lag {name child_interfaces {name}}}
          }
        }
        """,
        """
        connected_endpoints {
            __typename 
            ... on ProviderNetworkType {name}
            ... on InterfaceType {name, device {name, status}, child_interfaces {name}, lag {name child_interfaces {name}}}
        }
        """,
    ]
    interfaces_fields.append(
        """
        link_peers {
            __typename
            ... on InterfaceType {name device {name, status}}
            ... on FrontPortType {name device {name, status}}
            ... on RearPortType {name device {name, status}}
        }
    """
    )
    console_ports_fields = [
        "name",
        "device {name, status}",
        "type",
        """connected_endpoints {
          __typename 
          ... on ConsoleServerPortType {name device {name, status}}
        }""",
        """link_peers {
          __typename
          ... on ConsoleServerPortType {name device {name, status}}
          ... on FrontPortType {name device {name, status}}
          ... on RearPortType {name device {name, status}}
        }""",
    ]
    console_server_ports_fields = [
        "name",
        "device {name, status}",
        "type",
        """connected_endpoints {
          __typename 
          ... on ConsolePortType {name device {name, status}}
        }""",
        """link_peers {
          __typename
          ... on ConsolePortType {name device {name, status}}
          ... on FrontPortType {name device {name, status}}
          ... on RearPortType {name device {name, status}}
        }""",
    ]
    power_outlet_fields = [
        "name",
        "device {name, status}",
        "type",
        """connected_endpoints {
          __typename 
          ... on PowerPortType {name device {name, status}}
        }""",
        """link_peers {
          __typename
          ... on PowerPortType {name device {name, status}}
        }""",
    ]

    # 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)
        power_outlet_fields.append(cable_fields)

    # form query dictionary with aliases to get data from Netbox
    dlist = str(devices).replace("'", '"')  # swap quotes
    if self.nb_version[instance] >= (4, 4, 0):
        if interface_regex:
            filters = (
                "{device: {name: {in_list: "
                + dlist
                + "}}, "
                + "name: {i_regex: "
                + f'"{interface_regex}"'
                + "}}"
            )
        else:
            filters = "{device: {name: {in_list: " + dlist + "}}}"
    else:
        raise UnsupportedNetboxVersion(
            f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
            f"minimum required version is {self.compatible_ge_v4}"
        )

    queries = {
        "interface": {
            "obj": "interface_list",
            "filters": filters,
            "fields": interfaces_fields,
        },
        "consoleport": {
            "obj": "console_port_list",
            "filters": filters,
            "fields": console_ports_fields,
        },
        "consoleserverport": {
            "obj": "console_server_port_list",
            "filters": filters,
            "fields": console_server_ports_fields,
        },
        "poweroutlet": {
            "obj": "power_outlet_list",
            "filters": filters,
            "fields": power_outlet_fields,
        },
    }

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

    # return dry run result
    if dry_run:
        return query_result

    all_ports = query_result.result
    if not all_ports:
        return ret

    # extract physical interfaces connections
    for port_type, ports in all_ports.items():
        for port in ports:
            # skip ports that have no remote device connected
            endpoints = port["connected_endpoints"]
            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_device_status"] = 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"]
                connection["remote_device_status"] = endpoints[0]["device"][
                    "status"
                ]

            # 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

            # add physical connection to the results
            ret.result[device_name][port_name] = connection

    # extract virtual interfaces connections
    for port_type, ports in all_ports.items():
        for port in ports:
            # add child virtual interfaces connections
            if (
                not include_virtual
                or port["type"] != "virtual"
                or not port["parent"]
            ):
                continue
            device_name = port["device"]["name"]
            interface_name = port["name"]
            parent = port["parent"]
            connection = {
                "remote_device": None,
                "remote_device_status": None,
                "remote_interface": None,
                "remote_termination_type": "virtual",
                "termination_type": "virtual",
            }
            # find connection endpoint
            if parent["type"] == "lag":
                try:
                    endpoint = parent["member_interfaces"][0][
                        "connected_endpoints"
                    ][0]
                except:
                    continue
            elif parent["connected_endpoints"]:
                try:
                    endpoint = parent["connected_endpoints"][0]
                except:
                    continue
            connection["remote_device"] = endpoint["device"]["name"]
            connection["remote_device_status"] = endpoint["device"]["status"]
            remote_termination_type = endpoint["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")
            # collect virtual interfaces facing provider
            if remote_termination_type == "providernetwork":
                connection["provider"] = endpoint["name"]
            # find matching remote virtual interface for LAG subif
            elif "." in interface_name and parent["type"] == "lag":
                subif_id = interface_name.split(".")[1]
                for remote_child in endpoint["lag"]["child_interfaces"]:
                    if remote_child["name"].endswith(f".{subif_id}"):
                        connection["remote_interface"] = remote_child["name"]
                        break
                # no matching subinterface found, associate child interface with remote interface
                else:
                    connection["remote_interface"] = endpoint["lag"]["name"]
                    connection["remote_termination_type"] = "lag"
            # find matching remote virtual interface for physical interface subif
            elif "." in interface_name:
                subif_id = interface_name.split(".")[1]
                for remote_child in endpoint["child_interfaces"]:
                    if remote_child["name"].endswith(f".{subif_id}"):
                        connection["remote_interface"] = remote_child["name"]
                        break
                # no matching subinterface found, associate child interface with remote interface
                else:
                    connection["remote_interface"] = endpoint["name"]
                    connection["remote_termination_type"] = remote_termination_type
            # add virtual interface connection to results
            ret.result[device_name][interface_name] = connection

    # extract LAG interfaces connections
    for port_type, ports in all_ports.items():
        for port in ports:
            if not include_virtual or port["type"] != "lag":
                continue
            device_name = port["device"]["name"]
            interface_name = port["name"]
            connection = {
                "remote_device": None,
                "remote_device_status": None,
                "remote_interface": None,
                "remote_termination_type": "lag",
                "termination_type": "lag",
            }
            try:
                endpoint = port["member_interfaces"][0]["connected_endpoints"][0]
            except:
                continue
            remote_termination_type = endpoint["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")
            # collect lag interfaces facing provider
            if remote_termination_type == "providernetwork":
                connection["provider"] = endpoint["name"]
            # find remote lag interface
            elif endpoint["lag"]:
                connection["remote_interface"] = endpoint["lag"]["name"]
                connection["remote_device"] = endpoint["device"]["name"]
                connection["remote_device_status"] = endpoint["device"]["status"]
            # add lag interface connection to results
            ret.result[device_name][interface_name] = connection

    return ret

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

Retrieve circuit information for specified devices from Netbox.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
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	Result	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

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_circuits(
    self,
    job: Job,
    devices: list,
    cid: Union[None, list] = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    cache: Union[bool, str] = True,
) -> Result:
    """
    Retrieve circuit information for specified devices from Netbox.

    Args:
        job: NorFab Job object containing relevant metadata
        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.
    """
    cid = cid or []
    log.info(
        f"{self.name}:get_circuits - {instance or self.default_instance} Netbox, "
        f"devices {', '.join(devices)}, cid {cid}"
    )
    instance = instance or self.default_instance

    # form final result object
    ret = Result(
        task=f"{self.name}:get_circuits",
        result={d: {} for d in devices},
        resources=[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 = {}
    device_data = self.get_devices(
        job=job, 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] >= (4, 4, 0):
        slist = str(sites).replace("'", '"')  # swap quotes
        if cid:
            clist = str(cid).replace("'", '"')  # swap quotes
            circuits_filters = "{terminations: {site: {slug: {in_list: slist}}}, cid: {in_list: clist}}"
            circuits_filters = circuits_filters.replace("slist", slist).replace(
                "clist", clist
            )
        else:
            circuits_filters = "{terminations: {site: {slug: {in_list: slist }}}}"
            circuits_filters = circuits_filters.replace("slist", slist)
    else:
        raise UnsupportedNetboxVersion(
            f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
            f"minimum required version is {self.compatible_ge_v4}"
        )

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

    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(
            job=job,
            obj="circuit_list",
            filters=circuits_filters,
            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 = {}
        if cid_list:
            cid_list = str(cid_list).replace("'", '"')  # swap quotes
            if self.nb_version[instance] >= (4, 4, 0):
                circuits_filters = "{cid: {in_list: cid_list}}"
                circuits_filters = circuits_filters.replace("cid_list", cid_list)
            else:
                raise UnsupportedNetboxVersion(
                    f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )
    # ignore cache data, fetch circuits from netbox
    elif cache == False or cache == "refresh":
        pass

    if circuits_filters:
        query_result = self.graphql(
            job=job,
            obj="circuit_list",
            filters=circuits_filters,
            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, job, circuit, ret, instance, devices, cache
                )
                for circuit in all_circuits
            ]
            for _ in concurrent.futures.as_completed(results):
                continue

    return ret

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

Retrieve and construct Nornir inventory from NetBox data.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
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	Result	Nornir inventory dictionary containing hosts and their respective data.

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_nornir_inventory(
    self,
    job: Job,
    filters: Union[None, list] = None,
    devices: Union[None, list] = None,
    instance: Union[None, str] = None,
    interfaces: Union[dict, bool] = False,
    connections: Union[dict, bool] = False,
    circuits: Union[dict, bool] = False,
    nbdata: bool = True,
    bgp_peerings: Union[dict, bool] = False,
    primary_ip: str = "ip4",
) -> Result:
    """
    Retrieve and construct Nornir inventory from NetBox data.

    Args:
        job: NorFab Job object containing relevant metadata
        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 = {}
    filters = filters or []
    devices = devices or []
    inventory = {"hosts": hosts}
    ret = Result(task=f"{self.name}:get_nornir_inventory", result=inventory)

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

    # retrieve devices data
    nb_devices = self.get_devices(
        job=job, 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(
            job=job, 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(
            job=job, 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(
            job=job, 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

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

    return ret

get_nornir_hosts(kwargs: dict, timeout: int) -> List[str]

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	List[str]	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) -> List[str]:
    """
    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 and isinstance(r["result"], list):
                ret.extend(r["result"])

    return list(sorted(set(ret)))

sync_device_facts(job: Job, instance: Union[None, str] = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: Union[None, list] = None, batch_size: int = 10, branch: str = None, **kwargs: Any) -> Result

Updates device facts in NetBox, this task updates this device attributes:

• serial number

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
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
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None
**kwargs	Any	Additional keyword arguments to pass to the datasource job.	{}

Returns:

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

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def sync_device_facts(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: Union[None, list] = None,
    batch_size: int = 10,
    branch: str = None,
    **kwargs: Any,
) -> Result:
    """
    Updates device facts in NetBox, this task updates this device attributes:

    - serial number

    Args:
        job: NorFab Job object containing relevant metadata
        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.
        branch (str, optional): Branch name to use, need to have branching plugin installed,
            automatically creates branch if it does not exist in Netbox.
        **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.
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_facts",
        resources=[instance],
        dry_run=dry_run,
        diff={},
        result={},
    )
    nb = self._get_pynetbox(instance, branch=branch)
    kwargs["add_details"] = True

    if datasource == "nornir":
        # source hosts list from Nornir
        if kwargs:
            devices.extend(self.get_nornir_hosts(kwargs, timeout))
            devices = list(set(devices))
            job.event(f"Syncing {len(devices)} devices")
        # fetch devices data from Netbox
        nb_devices = self.get_devices(
            job=job,
            instance=instance,
            devices=copy.copy(devices),
            cache="refresh",
        ).result
        # remove devices that does not exist in Netbox
        for d in list(devices):
            if d not in nb_devices:
                msg = f"'{d}' device does not exist in Netbox"
                ret.errors.append(msg)
                log.error(msg)
                devices.remove(d)
        # 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"
            job.event(f"retrieving facts for devices {', '.join(kwargs['FL'])}")
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )

            # Collect devices to update in bulk
            devices_to_update = []

            for worker, results in data.items():
                if results["failed"]:
                    msg = f"{worker} get_facts failed, errors: {'; '.join(results['errors'])}"
                    ret.errors.append(msg)
                    log.error(msg)
                    continue
                for host, host_data in results["result"].items():
                    if host_data["napalm_get"]["failed"]:
                        msg = f"{host} facts update failed: '{host_data['napalm_get']['exception']}'"
                        ret.errors.append(msg)
                        log.error(msg)
                        continue

                    nb_device = nb_devices[host]

                    facts = host_data["napalm_get"]["result"]["get_facts"]
                    desired_state = {
                        "serial": facts["serial_number"],
                    }
                    current_state = {
                        "serial": nb_device["serial"],
                    }

                    # Compare and get fields that need updating
                    updates, diff = compare_netbox_object_state(
                        desired_state=desired_state,
                        current_state=current_state,
                    )

                    # Only update if there are changes
                    if updates:
                        updates["id"] = int(nb_device["id"])
                        devices_to_update.append(updates)
                        ret.diff[host] = diff

                    ret.result[host] = {
                        (
                            "sync_device_facts_dry_run"
                            if dry_run
                            else "sync_device_facts"
                        ): (updates if updates else "Device facts in sync")
                    }
                    if branch is not None:
                        ret.result[host]["branch"] = branch

            # Perform bulk update
            if devices_to_update and not dry_run:
                try:
                    nb.dcim.devices.update(devices_to_update)
                except Exception as e:
                    ret.errors.append(f"Bulk update failed: {e}")
    else:
        raise UnsupportedServiceError(
            f"'{datasource}' datasource service not supported"
        )

    return ret

sync_device_interfaces(job: Job, instance: Union[None, str] = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: Union[None, list] = None, create: bool = True, batch_size: int = 10, branch: str = None, **kwargs: Any) -> Result

Update or create device interfaces in Netbox using devices interfaces data sourced via Nornir service parse task using NAPALM getter.

Interface parameters updated:

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

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
instance	str	The Netbox 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
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None
**kwargs	Any	Additional keyword arguments to pass to the datasource job.	{}

Returns:

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

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def sync_device_interfaces(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: Union[None, list] = None,
    create: bool = True,
    batch_size: int = 10,
    branch: str = None,
    **kwargs: Any,
) -> Result:
    """
    Update or create device interfaces in Netbox using devices interfaces
    data sourced via Nornir service `parse` task using NAPALM getter.

    Interface parameters updated:

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

    Args:
        job: NorFab Job object containing relevant metadata.
        instance (str, optional): The Netbox 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.
        branch (str, optional): Branch name to use, need to have branching plugin installed,
            automatically creates branch if it does not exist in Netbox.
        **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.
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_interfaces",
        result={},
        resources=[instance],
        dry_run=dry_run,
        diff={},
    )
    nb = self._get_pynetbox(instance, branch=branch)
    kwargs["add_details"] = True

    if datasource == "nornir":
        # source hosts list from Nornir
        if kwargs:
            devices.extend(self.get_nornir_hosts(kwargs, timeout))
            devices = list(set(devices))
            job.event(f"syncing {len(devices)} devices")

        # fetch devices interfaces data from Netbox
        nb_interfaces_data = self.get_interfaces(
            job=job,
            instance=instance,
            devices=copy.copy(devices),
            cache="refresh",
        ).result

        # fetch devices data from Netbox
        nb_devices_data = self.get_devices(
            job=job,
            instance=instance,
            devices=copy.copy(devices),
        ).result

        # 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"
            job.event(
                f"retrieving interfaces for devices {', '.join(kwargs['FL'])}"
            )
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )

            # Collect interfaces to update and create in bulk
            interfaces_to_update = []
            interfaces_to_create = []
            mac_addresses_to_create = []

            for worker, results in data.items():
                if results["failed"]:
                    msg = f"{worker} get_interfaces failed, errors: {'; '.join(results['errors'])}"
                    ret.errors.append(msg)
                    log.error(msg)
                    continue

                for host, host_data in results["result"].items():
                    if host_data["napalm_get"]["failed"]:
                        msg = f"{host} interfaces update failed: '{host_data['napalm_get']['exception']}'"
                        ret.errors.append(msg)
                        log.error(msg)
                        continue

                    nb_interfaces = nb_interfaces_data.get(host, {})
                    if not nb_interfaces:
                        msg = f"'{host}' has no interfaces in Netbox, skipping"
                        ret.errors.append(msg)
                        log.warning(msg)
                        continue

                    # Get device ID for creating new interfaces
                    nb_device = nb_devices_data.get(host)
                    if not nb_device:
                        msg = f"'{host}' does not exist in Netbox"
                        ret.errors.append(msg)
                        log.error(msg)
                        continue

                    interfaces = host_data["napalm_get"]["result"]["get_interfaces"]

                    sync_key = "sync_device_interfaces"
                    create_key = "created_device_interfaces"
                    if dry_run:
                        sync_key = "sync_device_interfaces_dry_run"
                        create_key = "created_device_interfaces_dry_run"
                    ret.result[host] = {
                        sync_key: {},
                        create_key: {},
                    }
                    if branch is not None:
                        ret.result[host]["branch"] = branch

                    # Process network device interfaces
                    for intf_name, interface_data in interfaces.items():
                        if intf_name in nb_interfaces:
                            # Interface exists - prepare update
                            nb_intf = nb_interfaces[intf_name]

                            # Build desired state
                            desired_state = {
                                "description": interface_data.get(
                                    "description", ""
                                ),
                                "enabled": interface_data.get("is_enabled", True),
                            }
                            if 10000 > interface_data.get("mtu", 0) > 0:
                                desired_state["mtu"] = interface_data["mtu"]
                            if interface_data.get("speed", 0) > 0:
                                desired_state["speed"] = (
                                    interface_data["speed"] * 1000
                                )

                            # Build current state
                            current_state = {
                                "description": nb_intf.get("description", ""),
                                "enabled": nb_intf.get("enabled", True),
                            }
                            if nb_intf.get("mtu"):
                                current_state["mtu"] = nb_intf["mtu"]
                            if nb_intf.get("speed"):
                                current_state["speed"] = nb_intf["speed"]

                            # Compare and get fields that need updating
                            updates, diff = compare_netbox_object_state(
                                desired_state=desired_state,
                                current_state=current_state,
                            )

                            # Only update if there are changes
                            if updates:
                                updates["id"] = int(nb_intf["id"])
                                interfaces_to_update.append(updates)
                                ret.diff.setdefault(host, {})[intf_name] = diff

                            ret.result[host][sync_key][intf_name] = (
                                updates if updates else "Interface in sync"
                            )

                            mac_address = (
                                interface_data.get("mac_address", "")
                                .strip()
                                .lower()
                            )
                            if mac_address and mac_address not in ["none", ""]:
                                # Check if MAC already exists
                                for nb_mac in nb_intf.get("mac_addresses") or []:
                                    if (
                                        nb_mac.get("mac_address", "").lower()
                                        == mac_address
                                    ):
                                        break
                                else:
                                    # Prepare MAC address for creation
                                    mac_addresses_to_create.append(
                                        {
                                            "mac_address": mac_address,
                                            "assigned_object_type": "dcim.interface",
                                            "assigned_object_id": int(
                                                nb_intf["id"]
                                            ),
                                        }
                                    )
                        elif create:
                            # Interface doesn't exist - prepare creation
                            new_intf = {
                                "name": intf_name,
                                "device": int(nb_device["id"]),
                                "type": "other",
                                "description": interface_data.get(
                                    "description", ""
                                ),
                                "enabled": interface_data.get("is_enabled", True),
                            }
                            if 10000 > interface_data.get("mtu", 0) > 0:
                                new_intf["mtu"] = interface_data["mtu"]
                            if interface_data.get("speed", 0) > 0:
                                new_intf["speed"] = interface_data["speed"] * 1000

                            mac_address = (
                                interface_data.get("mac_address", "")
                                .strip()
                                .lower()
                            )
                            if mac_address and mac_address not in ["none", ""]:
                                mac_addresses_to_create.append(
                                    {
                                        "mac_address": mac_address,
                                        "assigned_object_type": "dcim.interface",
                                        "assigned_object_id": int(nb_intf["id"]),
                                    }
                                )

                            interfaces_to_create.append(new_intf)
                            ret.result[host][create_key][intf_name] = new_intf

            # Perform bulk updates and creations
            if interfaces_to_update and not dry_run:
                try:
                    nb.dcim.interfaces.update(interfaces_to_update)
                    job.event(
                        f"Bulk updated {len(interfaces_to_update)} interfaces"
                    )
                except Exception as e:
                    msg = f"Bulk interface update failed: {e}"
                    ret.errors.append(msg)
                    log.error(msg)

            if interfaces_to_create and not dry_run:
                try:
                    _ = nb.dcim.interfaces.create(interfaces_to_create)
                    job.event(
                        f"Bulk created {len(interfaces_to_create)} interfaces"
                    )
                except Exception as e:
                    msg = f"Bulk interface creation failed: {e}"
                    ret.errors.append(msg)
                    log.error(msg)

            # Bulk create MAC addresses
            if mac_addresses_to_create and not dry_run:
                try:
                    nb.dcim.mac_addresses.create(mac_addresses_to_create)
                    job.event(
                        f"Bulk created {len(mac_addresses_to_create)} MAC addresses"
                    )
                except Exception as e:
                    msg = f"Bulk MAC address creation failed: {e}"
                    ret.errors.append(msg)
                    log.error(msg)

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

    return ret

update_interfaces_description(job: Job, devices: list, description_template: str = None, descriptions: dict = None, interfaces: Union[None, list] = None, interface_regex: Union[None, str] = None, instance: Union[None, str] = None, dry_run: bool = False, timeout: int = 60, branch: str = None) -> Result

Updates the description of interfaces for specified devices in NetBox.

This method retrieves interface connections for the given devices, renders new descriptions using a Jinja2 template, and updates the interface descriptions in NetBox accordingly.

Only interfaces, console ports and console server ports supported.

Jinja2 environment receives these context variables for description template rendering:

• device - pynetbox dcim.device object
• interface - pynetbox object - dcim/interface, dcip.consoleport, dcim.consoleserverport - depending on what kind of interface is that.
• remote_device - string
• remote_interface - string
• termination_type - string
• cable - dictionary of directly attached cable attributes:
  • type
  • status
  • tenant - dictionary of {name: tenant_name}
  • label
  • tags - list of {name: tag_name} dictionaries
  • custom_fields - dictionary with custom fields data
  • peer_termination_type
  • peer_device
  • peer_interface

Parameters:

Name	Type	Description	Default
job	Job	The job context for logging and event handling.	required
devices	list	List of device names to update interfaces for.	required
description_template	str	Jinja2 template string for the interface description. Can reference remote template using nf://path/to/template.txt.	None
descriptions	dict	Dictionary keyed by interface names with values being interface description strings	None
interfaces	Union[None, list]	Specific interfaces to update.	None
interface_regex	Union[None, str]	Regex pattern to filter interfaces.	None
instance	Union[None, str]	NetBox instance identifier.	None
dry_run	bool	If True, performs a dry run without saving changes.	False
timeout	int	Timeout for NetBox API requests.	60
branch	str	Branch name for NetBox instance.	None

Returns:

Name	Type	Description
Result	Result	An object containing the outcome of the update operation, including before and after descriptions.

Source code in norfab\workers\netbox_worker.py

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def update_interfaces_description(
    self,
    job: Job,
    devices: list,
    description_template: str = None,
    descriptions: dict = None,
    interfaces: Union[None, list] = None,
    interface_regex: Union[None, str] = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    timeout: int = 60,
    branch: str = None,
) -> Result:
    """
    Updates the description of interfaces for specified devices in NetBox.

    This method retrieves interface connections for the given devices, renders
    new descriptions using a Jinja2 template, and updates the interface descriptions
    in NetBox accordingly.

    Only interfaces, console ports and console server ports supported.

    Jinja2 environment receives these context variables for description template rendering:

    - device - pynetbox `dcim.device` object
    - interface - pynetbox object - `dcim/interface`, `dcip.consoleport`,
        `dcim.consoleserverport` - depending on what kind of interface is that.
    - remote_device - string
    - remote_interface - string
    - termination_type - string
    - cable - dictionary of directly attached cable attributes:
        - type
        - status
        - tenant - dictionary of `{name: tenant_name}`
        - label
        - tags - list of `{name: tag_name}` dictionaries
        - custom_fields - dictionary with custom fields data
        - peer_termination_type
        - peer_device
        - peer_interface

    Args:
        job (Job): The job context for logging and event handling.
        devices (list): List of device names to update interfaces for.
        description_template (str): Jinja2 template string for the interface description.
            Can reference remote template using `nf://path/to/template.txt`.
        descriptions (dict): Dictionary keyed by interface names with values being interface
            description strings
        interfaces (Union[None, list], optional): Specific interfaces to update.
        interface_regex (Union[None, str], optional): Regex pattern to filter interfaces.
        instance (Union[None, str], optional): NetBox instance identifier.
        dry_run (bool, optional): If True, performs a dry run without saving changes.
        timeout (int, optional): Timeout for NetBox API requests.
        branch (str, optional): Branch name for NetBox instance.

    Returns:
        Result: An object containing the outcome of the update operation, including
            before and after descriptions.
    """
    result = {}
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:update_interfaces_description",
        result=result,
        resources=[instance],
    )
    nb = self._get_pynetbox(instance, branch=branch)

    if description_template:
        # get list of all interfaces connections
        nb_connections = self.get_connections(
            job=job,
            devices=devices,
            interface_regex=interface_regex,
            instance=instance,
            include_virtual=True,
            cables=True,
        )
        # produce interfaces description and update it
        while nb_connections.result:
            device, device_connections = nb_connections.result.popitem()
            ret.result.setdefault(device, {})
            for interface, connection in device_connections.items():
                job.event(f"{device}:{interface} updating description")
                if connection["termination_type"] == "consoleport":
                    nb_interface = nb.dcim.console_ports.get(
                        device=device, name=interface
                    )
                elif connection["termination_type"] == "consoleserverport":
                    nb_interface = nb.dcim.console_server_ports.get(
                        device=device, name=interface
                    )
                elif connection["termination_type"] == "powerport":
                    nb_interface = nb.dcim.power_ports.get(
                        device=device, name=interface
                    )
                elif connection["termination_type"] == "poweroutlet":
                    nb_interface = nb.dcim.power_outlets.get(
                        device=device, name=interface
                    )
                else:
                    nb_interface = nb.dcim.interfaces.get(
                        device=device, name=interface
                    )
                nb_device = nb.dcim.devices.get(name=device)
                rendered_description = self.jinja2_render_templates(
                    templates=[description_template],
                    context={
                        "device": nb_device,
                        "interface": nb_interface,
                        **connection,
                    },
                )
                rendered_description = str(rendered_description).strip()
                ret.result[device][interface] = {
                    "-": str(nb_interface.description),
                    "+": rendered_description,
                }
                nb_interface.description = rendered_description
                if dry_run is False:
                    nb_interface.save()
    if descriptions:
        for device in devices:
            ret.result.setdefault(device, {})
            for interface, description in descriptions.items():
                nb_interface = nb.dcim.interfaces.get(name=interface, device=device)
                if nb_interface:
                    ret.result[device][interface] = {
                        "-": str(nb_interface.description),
                        "+": description,
                    }
                    nb_interface.description = description
                    if dry_run is False:
                        nb_interface.save()
    return ret

sync_device_ip(job: Job, instance: Union[None, str] = None, dry_run: bool = False, datasource: str = 'nornir', timeout: int = 60, devices: Union[None, list] = None, create: bool = True, batch_size: int = 10, branch: str = None, **kwargs: Any) -> Result

Update the IP addresses of devices in Netbox.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	str	The Netbox 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
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None
**kwargs	Any	Additional keyword arguments.	{}

Returns:

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

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def sync_device_ip(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: Union[None, list] = None,
    create: bool = True,
    batch_size: int = 10,
    branch: str = None,
    **kwargs: Any,
) -> Result:
    """
    Update the IP addresses of devices in Netbox.

    Args:
        job: NorFab Job object containing relevant metadata
        instance (str, optional): The Netbox 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.
        branch (str, optional): Branch name to use, need to have branching plugin installed,
            automatically creates branch if it does not exist in Netbox.
        **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}:sync_device_ip", result=result, resources=[instance]
    )
    nb = self._get_pynetbox(instance, branch=branch)

    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] = {
                        "sync_ip_dry_run" if dry_run else "sync_ip": updated,
                        "created_ip_dry_run" if dry_run else "created_ip": created,
                    }
                    if branch is not None:
                        result[host]["branch"] = branch
                    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)
                                        job.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
                                job.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
                                job.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

create_ip(job: Job, prefix: Union[str, dict], device: Union[None, str] = None, interface: Union[None, str] = None, description: Union[None, str] = None, vrf: Union[None, str] = None, tags: Union[None, list] = None, dns_name: Union[None, str] = None, tenant: Union[None, str] = None, comments: Union[None, str] = None, role: Union[None, str] = None, status: Union[None, str] = None, is_primary: Union[None, bool] = None, instance: Union[None, str] = None, dry_run: Union[None, bool] = False, branch: Union[None, str] = None, mask_len: Union[None, int] = None, create_peer_ip: Union[None, bool] = True) -> Result

Allocate the next available IP address from a given subnet.

This task finds or creates an IP address in NetBox, updates its metadata, optionally links it to a device/interface, and supports a dry run mode for previewing changes.

Parameters:

Name	Type	Description	Default
prefix	str	The prefix from which to allocate the IP address, could be: IPv4 prefix string e.g. 10.0.0.0/24 IPv6 prefix string e.g. 2001::/64 Prefix description string to filter by Dictionary with prefix filters to feed pynetbox get method e.g. {"prefix": "10.0.0.0/24", "site__name": "foo"}	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.	False
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None
mask_len	int	mask length to use for IP address on creation or to update existing IP address. On new IP address creation will create child subnet of mask_len within parent prefix, new subnet not created for existing IP addresses. mask_len argument ignored on dry run and ip allocated from parent prefix directly.	None
create_peer_ip	bool	If True creates IP address for link peer - remote device interface connected to requested device and interface	True

Returns:

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

Tasks execution follow these steps:

1. Tries to find an existing IP in NetBox matching the device/interface/description. If found, uses it; otherwise, proceeds to create a new IP.

2. If prefix is a string, determines if it’s an IP network or a description. Builds a filter dictionary for NetBox queries, optionally including VRF.

3. Queries NetBox for the prefix using the constructed filter.

4. If dry_run is True, fetches the next available IP but doesn’t create it.

5. If not a dry run, creates the next available IP in the prefix.

6. Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags) if provided and different from current values. Handles interface assignment and can set the IP as primary for the device.

7. If changes were made and not a dry run, saves the IP and device updates to NetBox.

Source code in norfab\workers\netbox_worker.py

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def create_ip(
    self,
    job: Job,
    prefix: Union[str, dict],
    device: Union[None, str] = None,
    interface: Union[None, str] = None,
    description: Union[None, str] = None,
    vrf: Union[None, str] = None,
    tags: Union[None, list] = None,
    dns_name: Union[None, str] = None,
    tenant: Union[None, str] = None,
    comments: Union[None, str] = None,
    role: Union[None, str] = None,
    status: Union[None, str] = None,
    is_primary: Union[None, bool] = None,
    instance: Union[None, str] = None,
    dry_run: Union[None, bool] = False,
    branch: Union[None, str] = None,
    mask_len: Union[None, int] = None,
    create_peer_ip: Union[None, bool] = True,
) -> Result:
    """
    Allocate the next available IP address from a given subnet.

    This task finds or creates an IP address in NetBox, updates its metadata,
    optionally links it to a device/interface, and supports a dry run mode for
    previewing changes.

    Args:
        prefix (str): The prefix from which to allocate the IP address, could be:

            - IPv4 prefix string e.g. 10.0.0.0/24
            - IPv6 prefix string e.g. 2001::/64
            - Prefix description string to filter by
            - Dictionary with prefix filters to feed `pynetbox` get method
                e.g. `{"prefix": "10.0.0.0/24", "site__name": "foo"}`

        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.
        branch (str, optional): Branch name to use, need to have branching plugin
            installed, automatically creates branch if it does not exist in Netbox.
        mask_len (int, optional): mask length to use for IP address on creation or to
            update existing IP address. On new IP address creation will create child
            subnet of `mask_len` within parent `prefix`, new subnet not created for
            existing IP addresses. `mask_len` argument ignored on dry run and ip allocated
            from parent prefix directly.
        create_peer_ip (bool, optional): If True creates IP address for link peer -
            remote device interface connected to requested device and interface

    Returns:
        dict: A dictionary containing the result of the IP allocation.

    Tasks execution follow these steps:

    1. Tries to find an existing IP in NetBox matching the device/interface/description.
        If found, uses it; otherwise, proceeds to create a new IP.

    2. If prefix is a string, determines if it’s an IP network or a description.
        Builds a filter dictionary for NetBox queries, optionally including VRF.

    3. Queries NetBox for the prefix using the constructed filter.

    4. If dry_run is True, fetches the next available IP but doesn’t create it.

    5. If not a dry run, creates the next available IP in the prefix.

    6. Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags)
        if provided and different from current values. Handles interface assignment and
        can set the IP as primary for the device.

    7. If changes were made and not a dry run, saves the IP and device updates to NetBox.
    """
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:create_ip", result={}, resources=[instance])
    tags = tags or []
    has_changes = False
    nb_ip = None
    nb_device = None
    create_peer_ip_data = {}
    nb = self._get_pynetbox(instance, branch=branch)

    # source parent prefix from Netbox
    if isinstance(prefix, str):
        # try converting prefix to network, if fails prefix is not an IP network
        try:
            _ = ipaddress.ip_network(prefix)
            is_network = True
        except:
            is_network = False
        if is_network is True and vrf:
            prefix = {"prefix": prefix, "vrf__name": vrf}
        elif is_network is True:
            prefix = {"prefix": prefix}
        elif is_network is False and vrf:
            prefix = {"description": prefix, "vrf__name": vrf}
        elif is_network is False:
            prefix = {"description": prefix}
    nb_prefix = nb.ipam.prefixes.get(**prefix)
    if not nb_prefix:
        raise NetboxAllocationError(
            f"Unable to source parent prefix from Netbox - {prefix}"
        )
    parent_prefix_len = int(str(nb_prefix).split("/")[1])

    # try to source existing IP from netbox
    if device and interface and description:
        nb_ip = nb.ipam.ip_addresses.get(
            device=device,
            interface=interface,
            description=description,
            parent=str(nb_prefix),
        )
    elif device and interface:
        nb_ip = nb.ipam.ip_addresses.get(
            device=device, interface=interface, parent=str(nb_prefix)
        )
    elif description:
        nb_ip = nb.ipam.ip_addresses.get(
            description=description, parent=str(nb_prefix)
        )

    # create new IP address
    if not nb_ip:
        # check if interface has link peer that has IP within parent prefix
        if device and interface:
            connection = self.get_connections(
                job=job,
                devices=[device],
                interface_regex=interface,
                instance=instance,
                include_virtual=True,
            )
            if interface in connection.result[device]:
                peer = connection.result[device][interface]
                # do not process breakout cables
                if isinstance(peer["remote_interface"], list):
                    peer["remote_interface"] = None
                # try to source peer ip subnet
                nb_peer_ip = None
                if peer["remote_device"] and peer["remote_interface"]:
                    nb_peer_ip = nb.ipam.ip_addresses.get(
                        device=peer["remote_device"],
                        interface=peer["remote_interface"],
                        parent=str(nb_prefix),
                    )
                # try to source peer ip subnet
                nb_peer_prefix = None
                if nb_peer_ip:
                    peer_ip = ipaddress.ip_interface(nb_peer_ip.address)
                    nb_peer_prefix = nb.ipam.prefixes.get(
                        prefix=str(peer_ip.network),
                        vrf__name=vrf,
                    )
                elif create_peer_ip and peer["remote_interface"]:
                    create_peer_ip_data = {
                        "device": peer["remote_device"],
                        "interface": peer["remote_interface"],
                        "vrf": vrf,
                        "branch": branch,
                        "tenant": tenant,
                        "dry_run": dry_run,
                        "tags": tags,
                        "status": status,
                        "create_peer_ip": False,
                        "instance": instance,
                    }
                # use peer subnet to create IP address
                if nb_peer_prefix:
                    nb_prefix = nb_peer_prefix
                    mask_len = None  # cancel subnet creation
                    job.event(
                        f"Using link peer '{peer['remote_device']}:{peer['remote_interface']}' "
                        f"prefix '{nb_peer_prefix}' to create IP address"
                    )
        # if mask_len provided create new subnet
        if mask_len and not dry_run and mask_len != parent_prefix_len:
            if mask_len < parent_prefix_len:
                raise ValueError(
                    f"Mask length '{mask_len}' must be longer then '{parent_prefix_len}' prefix length"
                )
            prefix_status = status
            if prefix_status not in ["active", "reserved", "deprecated"]:
                prefix_status = None
            child_subnet = self.create_prefix(
                job=job,
                parent=str(nb_prefix),
                prefixlen=mask_len,
                vrf=vrf,
                tags=tags,
                tenant=tenant,
                status=prefix_status,
                instance=instance,
                branch=branch,
            )
            prefix = {"prefix": child_subnet.result["prefix"]}
            if vrf:
                prefix["vrf__name"] = vrf
            nb_prefix = nb.ipam.prefixes.get(**prefix)

            if not nb_prefix:
                raise NetboxAllocationError(
                    f"Unable to source child prefix of mask length "
                    f"'{mask_len}' from '{prefix}' parent prefix"
                )
        # execute dry run on new IP
        if dry_run is True:
            nb_ip = nb_prefix.available_ips.list()[0]
            ret.status = "unchanged"
            ret.dry_run = True
            ret.result = {
                "address": str(nb_ip),
                "description": description,
                "vrf": vrf,
                "device": device,
                "interface": interface,
            }
            # add branch to results
            if branch is not None:
                ret.result["branch"] = branch
            return ret
        # create new IP
        else:
            nb_ip = nb_prefix.available_ips.create()
            job.event(
                f"Created '{nb_ip}' IP address for '{device}:{interface}' within '{nb_prefix}' prefix"
            )
        ret.status = "created"
    else:
        job.event(f"Using existing IP address {nb_ip}")
        ret.status = "updated"

    # update IP address parameters
    if description and description != nb_ip.description:
        nb_ip.description = description
        has_changes = True
    if vrf and vrf != nb_ip.vrf:
        nb_ip.vrf = {"name": vrf}
        has_changes = True
    if tenant and tenant != nb_ip.tenant:
        nb_ip.tenant = {"name": tenant}
        has_changes = True
    if dns_name and dns_name != nb_ip.dns_name:
        nb_ip.dns_name = dns_name
        has_changes = True
    if comments and comments != nb_ip.comments:
        nb_ip.comments = comments
        has_changes = True
    if role and role != nb_ip.role:
        nb_ip.role = role
        has_changes = True
    if tags and not any(t in nb_ip.tags for t in tags):
        for t in tags:
            if t not in nb_ip.tags:
                nb_ip.tags.append({"name": t})
                has_changes = True
    if device and interface:
        nb_interface = nb.dcim.interfaces.get(device=device, name=interface)
        if not nb_interface:
            raise NetboxAllocationError(
                f"Unable to source '{device}:{interface}' interface from Netbox"
            )
        if (
            hasattr(nb_ip, "assigned_object")
            and nb_ip.assigned_object != nb_interface.id
        ):
            nb_ip.assigned_object_id = nb_interface.id
            nb_ip.assigned_object_type = "dcim.interface"
            if is_primary is not None:
                nb_device = nb.dcim.devices.get(name=device)
                nb_device.primary_ip4 = nb_ip.id
            has_changes = True
    if mask_len and not str(nb_ip).endswith(f"/{mask_len}"):
        address = str(nb_ip).split("/")[0]
        nb_ip.address = f"{address}/{mask_len}"
        has_changes = True

    # save IP address into Netbox
    if dry_run:
        ret.status = "unchanged"
        ret.dry_run = True
    elif has_changes:
        nb_ip.save()
        job.event(f"Updated '{str(nb_ip)}' IP address parameters")
        # make IP primary for device
        if is_primary is True and nb_device:
            nb_device.save()
    else:
        ret.status = "unchanged"

    # form and return results
    ret.result = {
        "address": str(nb_ip),
        "description": str(nb_ip.description),
        "vrf": str(nb_ip.vrf) if not vrf else nb_ip.vrf["name"],
        "device": device,
        "interface": interface,
    }
    # add branch to results
    if branch is not None:
        ret.result["branch"] = branch

    # create IP address for peer
    if create_peer_ip and create_peer_ip_data:
        job.event(
            f"Creating IP address for link peer '{create_peer_ip_data['device']}:{create_peer_ip_data['interface']}'"
        )
        peer_ip = self.create_ip(
            **create_peer_ip_data, prefix=str(nb_prefix), job=job
        )
        if peer_ip.failed == False:
            ret.result["peer"] = peer_ip.result

    return ret

create_prefix(job: Job, parent: Union[str, dict], description: str = None, prefixlen: int = 30, vrf: str = None, tags: Union[None, list] = None, tenant: str = None, comments: str = None, role: str = None, site: str = None, status: str = None, instance: Union[None, str] = None, dry_run: bool = False, branch: str = None) -> Result

Creates a new IP prefix in NetBox or updates an existing one.

Parameters:

Name	Type	Description	Default
parent	Union[str, dict]	Parent prefix to allocate new prefix from, could be: IPv4 prefix string e.g. 10.0.0.0/24 IPv6 prefix string e.g. 2001::/64 Prefix description string to filter by Dictionary with prefix filters for pynetbox prefixes.get method e.g. {"prefix": "10.0.0.0/24", "site__name": "foo"}	required
description	str	Description for the new prefix, prefix description used for deduplication to source existing prefixes.	None
prefixlen	int	The prefix length of the new prefix to create, by default allocates next available /30 point-to-point prefix.	30
vrf	str	Name of the VRF to associate with the prefix.	None
tags	Union[None, list]	List of tags to assign to the prefix.	None
tenant	str	Name of the tenant to associate with the prefix.	None
comments	str	Comments for the prefix.	None
role	str	Role to assign to the prefix.	None
site	str	Name of the site to associate with the prefix.	None
status	str	Status of the prefix.	None
instance	Union[None, str]	NetBox instance identifier.	None
dry_run	bool	If True, simulates the creation without making changes.	False
branch	str	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	None

Returns:

Name	Type	Description
Result	Result	An object containing the outcome, including status, details of the prefix, and resources used.

Source code in norfab\workers\netbox_worker.py

@Task(
    input=CreatePrefixInput,
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
)
def create_prefix(
    self,
    job: Job,
    parent: Union[str, dict],
    description: str = None,
    prefixlen: int = 30,
    vrf: str = None,
    tags: Union[None, list] = None,
    tenant: str = None,
    comments: str = None,
    role: str = None,
    site: str = None,
    status: str = None,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    branch: str = None,
) -> Result:
    """
    Creates a new IP prefix in NetBox or updates an existing one.

    Args:
        parent (Union[str, dict]): Parent prefix to allocate new prefix from, could be:

            - IPv4 prefix string e.g. 10.0.0.0/24
            - IPv6 prefix string e.g. 2001::/64
            - Prefix description string to filter by
            - Dictionary with prefix filters for `pynetbox` prefixes.get method
                e.g. `{"prefix": "10.0.0.0/24", "site__name": "foo"}`

        description (str): Description for the new prefix, prefix description used for
            deduplication to source existing prefixes.
        prefixlen (int, optional): The prefix length of the new prefix to create, by default
            allocates next available /30 point-to-point prefix.
        vrf (str, optional): Name of the VRF to associate with the prefix.
        tags (Union[None, list], optional): List of tags to assign to the prefix.
        tenant (str, optional): Name of the tenant to associate with the prefix.
        comments (str, optional): Comments for the prefix.
        role (str, optional): Role to assign to the prefix.
        site (str, optional): Name of the site to associate with the prefix.
        status (str, optional): Status of the prefix.
        instance (Union[None, str], optional): NetBox instance identifier.
        dry_run (bool, optional): If True, simulates the creation without making changes.
        branch (str, optional): Branch name to use, need to have branching plugin installed,
            automatically creates branch if it does not exist in Netbox.

    Returns:
        Result: An object containing the outcome, including status, details of the prefix, and resources used.
    """
    instance = instance or self.default_instance
    changed = {}
    ret = Result(
        task=f"{self.name}:create_prefix",
        result={},
        resources=[instance],
        diff=changed,
    )
    tags = tags or []
    nb_prefix = None
    nb = self._get_pynetbox(instance, branch=branch)

    job.event(
        f"Processing prefix create request within '{parent}' for '/{prefixlen}' subnet"
    )

    # source parent prefix from Netbox
    if isinstance(parent, str):
        # check if parent prefix is IP network or description
        try:
            _ = ipaddress.ip_network(parent)
            is_network = True
        except:
            is_network = False
        if is_network is True and vrf:
            parent_filters = {"prefix": parent, "vrf__name": vrf}
        elif is_network is True:
            parent_filters = {"prefix": parent}
        elif is_network is False and vrf:
            parent_filters = {"description": parent, "vrf__name": vrf}
        elif is_network is False:
            parent_filters = {"description": parent}
    nb_parent_prefix = nb.ipam.prefixes.get(**parent_filters)
    if not nb_parent_prefix:
        raise NetboxAllocationError(
            f"Unable to source parent prefix from Netbox - {parent}"
        )

    # check that parent vrf and new prefix vrf are same
    if vrf and str(nb_parent_prefix.vrf) != vrf:
        raise NetboxAllocationError(
            f"Parent prefix vrf '{nb_parent_prefix.vrf}' not same as requested child prefix vrf '{vrf}'"
        )

    # try to source existing prefix from netbox
    prefix_filters = {}
    if vrf:
        prefix_filters["vrf__name"] = vrf
    if site:
        prefix_filters["site__name"] = site
    if description:
        prefix_filters["description"] = description
    try:
        if prefix_filters:
            nb_prefix = nb.ipam.prefixes.get(
                within=nb_parent_prefix.prefix, **prefix_filters
            )
    except Exception as e:
        raise NetboxAllocationError(
            f"Failed to source existing prefix from Netbox using filters '{prefix_filters}', error: {e}"
        )

    # create new prefix
    if not nb_prefix:
        job.event(f"Creating new '/{prefixlen}' prefix within '{parent}' prefix")
        # execute dry run on new prefix
        if dry_run is True:
            nb_prefixes = nb_parent_prefix.available_prefixes.list()
            if not nb_prefixes:
                raise NetboxAllocationError(
                    f"Parent prefix '{parent}' has no child prefixes available"
                )
            for pfx in nb_prefixes:
                # parent prefix empty, can use first subnet as a child prefix
                if pfx.prefix == nb_parent_prefix.prefix:
                    nb_prefix = (
                        nb_parent_prefix.prefix.split("/")[0] + f"/{prefixlen}"
                    )
                    break
                # find child prefix by prefixlenght
                elif str(pfx).endswith(f"/{prefixlen}"):
                    nb_prefix = str(pfx)
                    break
            else:
                raise NetboxAllocationError(
                    f"Parent prefix '{parent}' has no child prefixes available with '/{prefixlen}' prefix length"
                )
            ret.status = "unchanged"
            ret.dry_run = True
            ret.result = {
                "prefix": nb_prefix,
                "description": description,
                "parent": nb_parent_prefix.prefix,
                "vrf": vrf,
                "site": site,
            }
            # add branch to results
            if branch is not None:
                ret.result["branch"] = branch
            return ret
        # create new prefix
        else:
            try:
                nb_prefix = nb_parent_prefix.available_prefixes.create(
                    {"prefix_length": prefixlen}
                )
            except Exception as e:
                raise NetboxAllocationError(
                    f"Failed creating child prefix of '/{prefixlen}' prefix length "
                    f"within parent prefix '{str(nb_parent_prefix)}', error: {e}"
                )
        job.event(f"Created new '{nb_prefix}' prefix within '{parent}' prefix")
        ret.status = "created"
    else:
        # check existing prefix length matching requested length
        if not nb_prefix.prefix.endswith(f"/{prefixlen}"):
            raise NetboxAllocationError(
                f"Found existing child prefix '{nb_prefix.prefix}' with mismatch "
                f"requested prefix length '/{prefixlen}'"
            )
        job.event(f"Using existing prefix {nb_prefix}")

    # update prefix parameters
    if description and description != nb_prefix.description:
        changed["description"] = {"-": str(nb_prefix.description), "+": description}
        nb_prefix.description = description
    if vrf and vrf != str(nb_prefix.vrf):
        changed["vrf"] = {"-": str(nb_prefix.vrf), "+": vrf}
        nb_prefix.vrf = {"name": vrf}
    if tenant and tenant != str(nb_prefix.tenant):
        changed["tenant"] = {
            "-": str(nb_prefix.tenant) if nb_prefix.tenant else None,
            "+": tenant,
        }
        nb_prefix.tenant = {"name": tenant}
    if site and str(nb_prefix.scope) != site:
        nb_site = nb.dcim.sites.get(name=site)
        if not nb_site:
            raise NetboxAllocationError(f"Failed to get '{site}' site from Netbox")
        changed["site"] = {
            "-": str(nb_prefix.scope) if nb_prefix.scope else None,
            "+": nb_site.name,
        }
        nb_prefix.scope_type = "dcim.site"
        nb_prefix.scope_id = nb_site.id
    if status and status.lower() != nb_prefix.status:
        changed["status"] = {"-": str(nb_prefix.status), "+": status.title()}
        nb_prefix.status = status.lower()
    if comments and comments != nb_prefix.comments:
        changed["comments"] = {"-": str(nb_prefix.comments), "+": comments}
        nb_prefix.comments = comments
    if role and role != nb_prefix.role:
        changed["role"] = {"-": str(nb_prefix.role), "+": role}
        nb_prefix.role = {"name": role}
    existing_tags = [str(t) for t in nb_prefix.tags]
    if tags and not any(t in existing_tags for t in tags):
        changed["tags"] = {
            "-": existing_tags,
            "+": [t for t in tags if t not in existing_tags] + existing_tags,
        }
        for t in tags:
            if t not in existing_tags:
                nb_prefix.tags.append({"name": t})

    # save prefix into Netbox
    if dry_run:
        ret.status = "unchanged"
        ret.dry_run = True
        ret.diff = changed
    elif changed:
        ret.diff = changed
        nb_prefix.save()
        if ret.status != "created":
            ret.status = "updated"
    else:
        ret.status = "unchanged"

    # source vrf name
    vrf_name = None
    if nb_prefix.vrf:
        if isinstance(nb_prefix.vrf, dict):
            vrf_name = nb_prefix.vrf["name"]
        else:
            vrf_name = nb_prefix.vrf.name

    # form and return results
    ret.result = {
        "prefix": nb_prefix.prefix,
        "description": nb_prefix.description,
        "vrf": vrf_name,
        "site": str(nb_prefix.scope) if nb_prefix.scope else site,
        "parent": nb_parent_prefix.prefix,
    }
    # add branch to results
    if branch is not None:
        ret.result["branch"] = branch

    return ret

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

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
job	Job	NorFab Job object containing relevant metadata	required
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
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	Result	Containerlab inventory dictionary containing lab topology data

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def get_containerlab_inventory(
    self,
    job: Job,
    lab_name: str = None,
    tenant: Union[None, str] = None,
    filters: Union[None, list] = None,
    devices: Union[None, list] = None,
    instance: Union[None, str] = None,
    image: Union[None, str] = None,
    ipv4_subnet: str = "172.100.100.0/24",
    ports: tuple = (12000, 15000),
    ports_map: Union[None, dict] = None,
    cache: Union[bool, str] = False,
) -> Result:
    """
    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:
        job: NorFab Job object containing relevant metadata
        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,
        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
    """
    devices = devices or []
    filters = filters or []
    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
    # add tenant filters
    if tenant:
        filters = filters or [{}]
        for filter in filters:
            if self.nb_version[instance] >= (4, 4, 0):
                filter["tenant"] = f'{{name: {{exact: "{tenant}"}}}}'
            else:
                raise UnsupportedNetboxVersion(
                    f"{self.name} - Netbox version {self.nb_version[instance]} is not supported, "
                    f"minimum required version is {self.compatible_ge_v4}"
                )

    # 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,
        resources=[instance],
    )
    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(job=job, 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}'"
    )
    job.event("Fetching devices data from Netbox")
    nb_devices = self.get_devices(
        job=job,
        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)
                job.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
        job.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

    job.event("Fetching connections data from Netbox")

    # query interface connections data from netbox
    nb_connections = self.get_connections(
        job=job, 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)
                job.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(
        job=job, 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)
                job.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:
            job.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)
                        job.event(msg)
                        endpoint["interface"] = renamed

    return ret

delete_branch(job: Job, branch: str = None, instance: str = None) -> Result

Deletes a branch with the specified name from the NetBox instance.

Parameters:

Name	Type	Description	Default
job	Job	The job context for the operation.	required
branch	str	The name of the branch to delete.	None
instance	str	The NetBox instance name.	None

Returns:

Name	Type	Description
Result	Result	An object containing the outcome of the deletion operation, including whether the branch was found and deleted.

Source code in norfab\workers\netbox_worker.py

@Task(
    fastapi={"methods": ["DELETE"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def delete_branch(
    self,
    job: Job,
    branch: str = None,
    instance: str = None,
) -> Result:
    """
    Deletes a branch with the specified name from the NetBox instance.

    Args:
        job (Job): The job context for the operation.
        branch (str, optional): The name of the branch to delete.
        instance (str, optional): The NetBox instance name.

    Returns:
        Result: An object containing the outcome of the deletion operation,
            including whether the branch was found and deleted.
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:delete_branch",
        result=None,
        resources=[instance],
    )
    nb = self._get_pynetbox(instance)

    job.event(f"Deleting branch '{branch}', Netbo instance '{instance}'")

    nb_branch = nb.plugins.branching.branches.get(name=branch)

    if nb_branch:
        nb_branch.delete()
        ret.result = True
        job.event(f"'{branch}' deleted from '{instance}' Netbox instance")
    else:
        msg = f"'{branch}' branch does not exist in '{instance}' Netbox instance"
        ret.result = None
        ret.messages.append(msg)
        job.event(msg)

    return ret

expand_alphanumeric_range(range_pattern: str) -> list

Expand alphanumeric ranges.

Examples:

• Ethernet[1-3] -> ['Ethernet1', 'Ethernet2', 'Ethernet3']
• [ge,xe]-0/0/[0-9] -> ['ge-0/0/0', 'ge-0/0/1', ..., 'xe-0/0/9']

Source code in norfab\workers\netbox_worker.py

def expand_alphanumeric_range(self, range_pattern: str) -> list:
    """
    Expand alphanumeric ranges.

    Examples:
        - Ethernet[1-3] -> ['Ethernet1', 'Ethernet2', 'Ethernet3']
        - [ge,xe]-0/0/[0-9] -> ['ge-0/0/0', 'ge-0/0/1', ..., 'xe-0/0/9']
    """
    # Find all bracketed patterns
    bracket_pattern = r"\[([^\]]+)\]"
    matches = list(re.finditer(bracket_pattern, range_pattern))

    if not matches:
        # No ranges found, return as-is
        return [range_pattern]

    # Start with a single template
    templates = [range_pattern]

    # Process each bracket from left to right
    for match in matches:
        bracket_content = match.group(1)
        new_templates = []

        # Check if it's a comma-separated list
        if "," in bracket_content:
            options = [opt.strip() for opt in bracket_content.split(",")]
            for template in templates:
                for option in options:
                    new_templates.append(
                        template.replace(f"[{bracket_content}]", option, 1)
                    )

        # Check if it's a numeric range
        elif (
            "-" in bracket_content
            and bracket_content.replace("-", "").replace(" ", "").isdigit()
        ):
            parts = bracket_content.split("-")
            if len(parts) == 2:
                try:
                    start = int(parts[0].strip())
                    end = int(parts[1].strip())
                    for template in templates:
                        for num in range(start, end + 1):
                            new_templates.append(
                                template.replace(
                                    f"[{bracket_content}]", str(num), 1
                                )
                            )
                except ValueError:
                    # If conversion fails, treat as literal
                    for template in templates:
                        new_templates.append(
                            template.replace(
                                f"[{bracket_content}]", bracket_content, 1
                            )
                        )
        else:
            # Treat as literal
            for template in templates:
                new_templates.append(
                    template.replace(f"[{bracket_content}]", bracket_content, 1)
                )

        templates = new_templates

    return templates

create_device_interfaces(job: Job, devices: list, interface_name: Union[list, str], interface_type: str = 'other', instance: Union[None, str] = None, dry_run: bool = False, branch: str = None, **kwargs: dict) -> Result

Create interfaces for one or more devices in NetBox. This task creates interfaces in bulk and only if interfaces does not exist in Netbox.

Parameters:

Name	Type	Description	Default
job	Job	The job object containing execution context and metadata.	required
devices	list	List of device names or device objects to create interfaces for.	required
interface_name	Union[list, str]	Name(s) of the interface(s) to create. Can be a single interface name as a string or multiple names as a list. Alphanumeric ranges are supported for bulk creation: Ethernet[1-3] -> Ethernet1, Ethernet2, Ethernet3 [ge,xe]-0/0/[0-9] -> ge-0/0/0, ..., xe-0/0/0 etc.	required
interface_type	str	Type of interface (e.g., "other", "virtual", "lag", "1000base-t"). Defaults to "other".	'other'
instance	Union[None, str]	NetBox instance identifier to use. If None, uses the default instance. Defaults to None.	None
dry_run	bool	If True, simulates the operation without making actual changes. Defaults to False.	False
branch	str	NetBox branch to use for the operation. Defaults to None.	None
kwargs	dict	Any additional interface attributes	{}

Returns:

Name	Type	Description
Result	Result	Result object containing the task name, execution results, and affected resources. The result dictionary contains status and details of interface creation operations.

Source code in norfab\workers\netbox_worker.py

@Task(fastapi={"methods": ["GET"], "schema": NetboxFastApiArgs.model_json_schema()})
def create_device_interfaces(
    self,
    job: Job,
    devices: list,
    interface_name: Union[list, str],
    interface_type: str = "other",
    instance: Union[None, str] = None,
    dry_run: bool = False,
    branch: str = None,
    **kwargs: dict,
) -> Result:
    """
    Create interfaces for one or more devices in NetBox. This task creates interfaces in bulk and only
    if interfaces does not exist in Netbox.

    Args:
        job (Job): The job object containing execution context and metadata.
        devices (list): List of device names or device objects to create interfaces for.
        interface_name (Union[list, str]): Name(s) of the interface(s) to create. Can be a single
            interface name as a string or multiple names as a list. Alphanumeric ranges are
            supported for bulk creation:

            - Ethernet[1-3] -> Ethernet1, Ethernet2, Ethernet3
            - [ge,xe]-0/0/[0-9] -> ge-0/0/0, ..., xe-0/0/0 etc.

        interface_type (str, optional): Type of interface (e.g., "other", "virtual", "lag",
            "1000base-t"). Defaults to "other".
        instance (Union[None, str], optional): NetBox instance identifier to use. If None,
            uses the default instance. Defaults to None.
        dry_run (bool, optional): If True, simulates the operation without making actual changes.
            Defaults to False.
        branch (str, optional): NetBox branch to use for the operation. Defaults to None.
        kwargs (dict, optional): Any additional interface attributes

    Returns:
        Result: Result object containing the task name, execution results, and affected resources.
            The result dictionary contains status and details of interface creation operations.
    """
    instance = instance or self.default_instance
    result = {}
    kwargs = kwargs or {}
    ret = Result(
        task=f"{self.name}:create_device_interfaces",
        result=result,
        resources=[instance],
    )
    nb = self._get_pynetbox(instance, branch=branch)

    # Normalize interface_name to a list
    if isinstance(interface_name, str):
        interface_names = [interface_name]
    else:
        interface_names = interface_name

    # Expand all interface name patterns
    all_interface_names = []
    for name_pattern in interface_names:
        all_interface_names.extend(self.expand_alphanumeric_range(name_pattern))

    job.event(
        f"Expanded interface names to {len(all_interface_names)} interface(s)"
    )

    # Process each device
    for device_name in devices:
        result[device_name] = {
            "created": [],
            "skipped": [],
        }

        try:
            # Get device from NetBox
            nb_device = nb.dcim.devices.get(name=device_name)
            if not nb_device:
                msg = f"Device '{device_name}' not found in NetBox"
                ret.errors.append(msg)
                job.event(msg)
                continue

            # Get existing interfaces for this device
            existing_interfaces = nb.dcim.interfaces.filter(device=device_name)
            existing_interface_names = {intf.name for intf in existing_interfaces}

            # Prepare interfaces to create
            interfaces_to_create = []

            for intf_name in all_interface_names:
                if intf_name in existing_interface_names:
                    result[device_name]["skipped"].append(intf_name)
                    continue

                # Build interface data
                intf_data = {
                    "device": nb_device.id,
                    "name": intf_name,
                    "type": interface_type,
                    **kwargs,
                }

                interfaces_to_create.append(intf_data)
                result[device_name]["created"].append(intf_name)

            # Create interfaces in bulk if not dry_run
            if interfaces_to_create and not dry_run:
                try:
                    nb.dcim.interfaces.create(interfaces_to_create)
                    msg = f"Created {len(interfaces_to_create)} interface(s) on device '{device_name}'"
                    job.event(msg)
                except Exception as e:
                    msg = f"Failed to create interfaces on device '{device_name}': {e}"
                    ret.errors.append(msg)
                    log.error(msg)
            elif interfaces_to_create and dry_run:
                msg = f"[DRY RUN] Would create {len(interfaces_to_create)} interface(s) on device '{device_name}'"
                job.event(msg)

        except Exception as e:
            msg = f"Error processing device '{device_name}': {e}"
            ret.errors.append(msg)
            log.error(msg)

    return ret

get_bgp_peerings(job: Job, instance: Union[None, str] = None, devices: Union[None, list] = None, cache: Union[bool, str] = None) -> Result

Retrieve device BGP peerings from Netbox using REST API.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata	required
instance	str	Netbox instance name.	None
devices	list	List of devices to retrieve BGP peerings for.	None
cache	Union[bool, str]	Cache usage options: True: Use data stored in cache if it is up to date, refresh it otherwise. False: Do not use cache and do not update cache. refresh: Ignore data in cache and replace it with data fetched from Netbox. force: Use data in cache without checking if it is up to date.	None

Returns:

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

Source code in norfab\workers\netbox_worker.py

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

    Args:
        job: NorFab Job object containing relevant metadata
        instance (str, optional): Netbox instance name.
        devices (list, optional): List of devices to retrieve BGP peerings 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: Dictionary keyed by device name with BGP peerings details.
    """
    instance = instance or self.default_instance
    devices = devices or []
    cache = self.cache_use if cache is None else cache
    ret = Result(
        task=f"{self.name}:get_bgp_peerings",
        result={d: {} for d in devices},
        resources=[instance],
    )

    # Check if BGP plugin is installed
    if not self.has_plugin("netbox_bgp", instance, strict=True):
        ret.errors.append(f"{instance} Netbox instance has no BGP Plugin installed")
        ret.failed = True
        return ret

    self.cache.expire()

    # Get device details to collect device IDs
    devices_result = self.get_devices(
        job=job, devices=devices, instance=instance, cache=False
    )
    if devices_result.errors:
        ret.errors.append(
            f"Failed to retrieve device details: {devices_result.errors}"
        )
        return ret

    nb = self._get_pynetbox(instance)

    for device_name in devices:
        # Skip devices not found in Netbox
        if device_name not in devices_result.result:
            msg = f"Device '{device_name}' not found in Netbox"
            job.event(msg, resource=instance, severity="WARNING")
            log.warning(msg)
            continue

        device_id = devices_result.result[device_name]["id"]
        cache_key = f"get_bgp_peerings::{device_name}"
        cached_data = self.cache.get(cache_key)

        # Mode: force with cached data - use cache directly
        if cache == "force" and cached_data is not None:
            ret.result[device_name] = cached_data
            job.event(
                f"Using cached BGP peerings for '{device_name}' (forced)",
                resource=instance,
            )
            continue

        # Mode: cache disabled - fetch without caching
        if cache is False:
            bgp_sessions = nb.plugins.bgp.session.filter(device_id=device_id)
            ret.result[device_name] = {s.name: dict(s) for s in bgp_sessions}
            job.event(
                f"Retrieved {len(ret.result[device_name])} BGP session(s) for '{device_name}'",
                resource=instance,
            )
            continue

        # Mode: refresh or no cached data - fetch and cache
        if cache == "refresh" or cached_data is None:
            if cache == "refresh" and cached_data is not None:
                self.cache.delete(cache_key, retry=True)
            bgp_sessions = nb.plugins.bgp.session.filter(device_id=device_id)
            ret.result[device_name] = {s.name: dict(s) for s in bgp_sessions}
            self.cache.set(
                cache_key, ret.result[device_name], expire=self.cache_ttl
            )
            job.event(
                f"Fetched and cached {len(ret.result[device_name])} BGP session(s) for '{device_name}'",
                resource=instance,
            )
            continue

        # Mode: cache=True with cached data - smart update (only fetch changed sessions)
        ret.result[device_name] = dict(cached_data)
        job.event(
            f"Retrieved {len(cached_data)} BGP session(s) from cache for '{device_name}'",
            resource=instance,
        )

        # Fetch brief session info to compare timestamps
        brief_sessions = nb.plugins.bgp.session.filter(
            device_id=device_id, fields="id,last_updated,name"
        )
        netbox_sessions = {
            s.id: {"name": s.name, "last_updated": s.last_updated}
            for s in brief_sessions
        }

        # Build lookup maps
        cached_by_id = {s["id"]: name for name, s in cached_data.items()}
        session_ids_to_fetch = []
        sessions_to_remove = []

        # Find stale sessions (exist in both but timestamps differ) and deleted sessions
        for session_name, cached_session in cached_data.items():
            cached_id = cached_session["id"]
            if cached_id in netbox_sessions:
                if (
                    cached_session["last_updated"]
                    != netbox_sessions[cached_id]["last_updated"]
                ):
                    session_ids_to_fetch.append(cached_id)
            else:
                sessions_to_remove.append(session_name)

        # Find new sessions in Netbox not in cache
        for nb_id in netbox_sessions:
            if nb_id not in cached_by_id:
                session_ids_to_fetch.append(nb_id)

        # Remove deleted sessions
        for session_name in sessions_to_remove:
            ret.result[device_name].pop(session_name, None)
            job.event(
                f"Removed deleted session '{session_name}' from cache for '{device_name}'",
                resource=instance,
            )

        # Fetch updated/new sessions
        if session_ids_to_fetch:
            job.event(
                f"Fetching {len(session_ids_to_fetch)} updated BGP session(s) for '{device_name}'",
                resource=instance,
            )
            for session in nb.plugins.bgp.session.filter(id=session_ids_to_fetch):
                ret.result[device_name][session.name] = dict(session)

        # Update cache if any changes occurred
        if session_ids_to_fetch or sessions_to_remove:
            self.cache.set(
                cache_key, ret.result[device_name], expire=self.cache_ttl
            )
            job.event(f"Updated cache for '{device_name}'", resource=instance)
        else:
            job.event(
                f"Using cache, it is up to date for '{device_name}'",
                resource=instance,
            )

    return ret

compare_netbox_object_state(desired_state: dict, current_state: dict, ignore_fields: Union[list, None] = None, ignore_if_not_empty: Union[list, None] = None, diff: dict = None) -> tuple

Compare desired state with current NetBox object state and return fields that need updating.

Parameters:

Name	Type	Description	Default
desired_state	dict	Dictionary with desired field values.	required
current_state	dict	Dictionary with current NetBox object field values.	required
ignore_fields	list	List of field names to ignore completely.	None
ignore_if_not_empty	list	List of field names to ignore if they have non-empty values in current_state (won't overwrite existing data).	None
diff	dict	Dictionary to accumulate field differences. If not provided, a new dictionary will be created.	None

Returns:

Name	Type	Description
tuple	tuple	A tuple containing: - updates (dict): Dictionary containing only fields that need to be updated with their new values. - diff (dict): Dictionary containing the differences with '+' (new value) and '-' (old value) keys.

Example

desired = {"serial": "ABC123", "asset_tag": "TAG001", "comments": "New comment"} current = {"serial": "OLD123", "asset_tag": "TAG001", "comments": "Existing"} ignore_fields = ["comments"] ignore_if_not_empty = [] updates, diff = compare_netbox_object_state(desired, current, ignore_fields, ignore_if_not_empty) updates

desired = {"serial": "ABC123", "asset_tag": "TAG001", "comments": "New comment"} current = {"serial": "OLD123", "asset_tag": "", "comments": "Existing"} ignore_fields = [] ignore_if_not_empty = ["comments"] updates, diff = compare_netbox_object_state(desired, current, ignore_fields, ignore_if_not_empty) updates

Source code in norfab\workers\netbox_worker.py

def compare_netbox_object_state(
    desired_state: dict,
    current_state: dict,
    ignore_fields: Union[list, None] = None,
    ignore_if_not_empty: Union[list, None] = None,
    diff: dict = None,
) -> tuple:
    """
    Compare desired state with current NetBox object state and return fields that need updating.

    Args:
        desired_state (dict): Dictionary with desired field values.
        current_state (dict): Dictionary with current NetBox object field values.
        ignore_fields (list, optional): List of field names to ignore completely.
        ignore_if_not_empty (list, optional): List of field names to ignore if they have
            non-empty values in current_state (won't overwrite existing data).
        diff (dict, optional): Dictionary to accumulate field differences. If not provided,
            a new dictionary will be created.

    Returns:
        tuple: A tuple containing:
            - updates (dict): Dictionary containing only fields that need to be updated with their new values.
            - diff (dict): Dictionary containing the differences with '+' (new value) and '-' (old value) keys.

    Example:
        >>> desired = {"serial": "ABC123", "asset_tag": "TAG001", "comments": "New comment"}
        >>> current = {"serial": "OLD123", "asset_tag": "TAG001", "comments": "Existing"}
        >>> ignore_fields = ["comments"]
        >>> ignore_if_not_empty = []
        >>> updates, diff = compare_netbox_object_state(desired, current, ignore_fields, ignore_if_not_empty)
        >>> updates
        {"serial": "ABC123"}

        >>> desired = {"serial": "ABC123", "asset_tag": "TAG001", "comments": "New comment"}
        >>> current = {"serial": "OLD123", "asset_tag": "", "comments": "Existing"}
        >>> ignore_fields = []
        >>> ignore_if_not_empty = ["comments"]
        >>> updates, diff = compare_netbox_object_state(desired, current, ignore_fields, ignore_if_not_empty)
        >>> updates
        {"serial": "ABC123", "asset_tag": "TAG001"}
    """
    ignore_fields = ignore_fields or []
    ignore_if_not_empty = ignore_if_not_empty or []
    updates = {}
    diff = diff or {}

    for field, desired_value in desired_state.items():
        # Skip if field is in ignore list
        if field in ignore_fields:
            continue

        # Get current value, default to None if field doesn't exist
        current_value = current_state.get(field)

        # Skip if field is in ignore_if_not_empty and current value is not empty
        if field in ignore_if_not_empty and current_value:
            continue

        # Compare values and add to updates if different
        if current_value != desired_value:
            updates[field] = desired_value
            diff[field] = {
                "-": current_value,
                "+": desired_value,
            }

    return updates, diff