Netbox Get Circuits Task

task api name: get_circuits

How It Works

Sample devices' circuits data retrieved from Netbox:

{
    "netbox-worker-1.1": {
        "fceos4": {
            "CID1": {
                "comments": "",
                "commit_rate": null,
                "custom_fields": {},
                "description": "",
                "interface": "eth101",
                "is_active": true,
                "last_updated": "2026-01-02T22:50:14.739796+00:00",
                "provider": "Provider1",
                "provider_account": "",
                "remote_device": "fceos5",
                "remote_interface": "eth101",
                "status": "active",
                "tags": [],
                "tenant": null,
                "termination_a": {
                    "id": "36",
                    "last_updated": "2026-01-02T22:50:12.085037+00:00"
                },
                "termination_z": {
                    "id": "37",
                    "last_updated": "2026-01-02T22:50:14.498313+00:00"
                },
                "type": "DarkFibre"
            },
            "CID2": {
              ... etc ...

NORFAB Netbox Get Circuits Command Shell Reference

NorFab shell supports these command options for Netbox get_circuits task:

nf#man tree netbox.get.circuits
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── circuits:    Query Netbox circuits data for devices
            ├── instance:    Netbox instance name to target
            ├── workers:    Filter worker to target, default 'any'
            ├── timeout:    Job timeout
            ├── device-list:    Device names to query data for
            ├── dry-run:    Only return query content, do not run it
            ├── cid:    List of circuit identifiers to retrieve data for
            └── cache:    How to use cache, default 'True'
nf#

Python API Reference

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