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.

How it works: 1. User requests circuits for devices: ["device1", "device2"] 2. Fetch device data → Extract their sites: ["site1", "site2"] 3. Query circuits WHERE terminations.site IN (site1, site2) ← Broad query! 4. For each circuit, trace termination paths 5. Map terminations to specific requested devices 6. Filter results client-side

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
add_interface_details	bool	If True, add interface details using get_interfaces call including interface subinterfaces - ip addresses, vrf, child interfaces with their IPs and vrf	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.	None

Returns:

Name	Type	Description
dict	Result	dictionary keyed by device names with circuits data: nf#netbox get circuits device-list fceos5 { "netbox-worker-1.1": { "fceos5": { "CID1": { "comments": "", "commit_rate": null, "custom_fields": {}, "description": "", "interface": "eth101", "is_active": true, "provider": "Provider1", "provider_account": "", "remote_device": "fceos4", "remote_interface": "eth101", "status": "active", "tags": [], "tenant": null, "type": "DarkFibre" } } } } nf#

If add_interface_details is True returns this extra information:

{
    "netbox-worker-1.1": {
        "fceos4": {
            "CID2": {
                "child_interfaces": [
                    {
                        "ip_addresses": [
                            {
                                "address": "10.0.0.12/24",
                                ...
                            }
                        ],
                        "name": "eth11.123",
                        "vrf": {
                            "name": "MGMT"
                        }
                    }
                ],
                "interface": "eth11",
                "ip_addresses": [
                    {
                        "address": "10.0.0.14/24",
                        ...
                    }
                ],
                "vrf": {
                    "name": "OOB_CTRL"
                }
            }
        }
    }
}

Source code in norfab\workers\netbox_worker\circuits_tasks.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[None, bool, str] = None,
    add_interface_details: bool = False,
) -> Result:
    """
    Retrieve circuit information for specified devices from Netbox.

    **How it works:**
    1. User requests circuits for devices: `["device1", "device2"]`
    2. Fetch device data → Extract their sites: `["site1", "site2"]`
    3. Query circuits WHERE `terminations.site IN (site1, site2)` ← **Broad query!**
    4. For each circuit, trace termination paths
    5. Map terminations to specific requested devices
    6. Filter results client-side

    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.
        add_interface_details (bool, optional): If True, add interface details using `get_interfaces` call
            including interface subinterfaces - ip addresses, vrf, child interfaces with their IPs and vrf
        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:

            ```
            nf#netbox get circuits device-list fceos5
            {
                "netbox-worker-1.1": {
                    "fceos5": {
                        "CID1": {
                            "comments": "",
                            "commit_rate": null,
                            "custom_fields": {},
                            "description": "",
                            "interface": "eth101",
                            "is_active": true,
                            "provider": "Provider1",
                            "provider_account": "",
                            "remote_device": "fceos4",
                            "remote_interface": "eth101",
                            "status": "active",
                            "tags": [],
                            "tenant": null,
                            "type": "DarkFibre"
                        }
                    }
                }
            }
            nf#
            ```

    If `add_interface_details` is True returns this extra information:

    ```
    {
        "netbox-worker-1.1": {
            "fceos4": {
                "CID2": {
                    "child_interfaces": [
                        {
                            "ip_addresses": [
                                {
                                    "address": "10.0.0.12/24",
                                    ...
                                }
                            ],
                            "name": "eth11.123",
                            "vrf": {
                                "name": "MGMT"
                            }
                        }
                    ],
                    "interface": "eth11",
                    "ip_addresses": [
                        {
                            "address": "10.0.0.14/24",
                            ...
                        }
                    ],
                    "vrf": {
                        "name": "OOB_CTRL"
                    }
                }
            }
        }
    }
    ```
    """
    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")
        job.event("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:
        job.event("fetching circuits data from Netbox")
        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
        msg = (
            f"retrieved data for {len(all_circuits)} "
            f"circuits from Netbox, mapping circuits to devices"
        )
        log.info(msg)
        job.event(msg)
        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

    if add_interface_details:
        job.event("fetching circuits interface details")
        # collect devices and interfaces to get details for
        fetch_interfaces = set()
        fetch_devices = set(ret.result.keys())
        for device_name, circuits in ret.result.items():
            for ckt_data in circuits.values():
                fetch_interfaces.add(ckt_data["interface"])
                if ckt_data.get("remote_device"):
                    fetch_devices.add(ckt_data["remote_device"])
                    fetch_interfaces.add(ckt_data["remote_interface"])
        # fetch interfaces data with IP addresses
        interfaces_data = self.get_interfaces(
            job=job,
            devices=list(fetch_devices),
            interface_list=list(fetch_interfaces),
            ip_addresses=True,
        ).result
        # map interfaces details to circuits
        for device_name, circuits in ret.result.items():
            for circuit_id, ckt_data in circuits.items():
                interface_name = ckt_data["interface"]
                if interface_name in interfaces_data.get(device_name, {}):
                    interface_data = interfaces_data[device_name][interface_name]
                    ckt_data["child_interfaces"] = interface_data.get(
                        "child_interfaces", []
                    )
                    ckt_data["ip_addresses"] = interface_data.get(
                        "ip_addresses", []
                    )
                    ckt_data["vrf"] = interface_data.get("vrf", {})
                else:
                    log.error(
                        f"{device_name}:{circuit_id} Failed to find '{interface_name}' interface details"
                    )

    return ret