Netbox Check Device Sync Task

task api name: check_device_sync

Checks whether NetBox data is in sync with live device state without writing to NetBox. The task runs selected sync tasks in dry_run=True mode and returns a per-device summary plus detailed dry-run diffs.

How It Works

check_device_sync can run these read-only sub-checks:

• interfaces — calls sync_device_interfaces(dry_run=True)
• mac_addresses — calls sync_mac_addresses(dry_run=True)
• ip_addresses — calls sync_device_ip(dry_run=True)
• bgp_peerings — calls sync_bgp_peerings(dry_run=True)

Each sub-check can be enabled or disabled independently.

Inputs

Parameter	Required	Description
devices	No	NetBox device names to check
instance	No	NetBox instance name to target
branch	No	NetBox Branching plugin branch name to read from
timeout	No	Timeout in seconds for Nornir parse jobs
check_interfaces	No	Check interface sync state, default True
check_mac_addresses	No	Check MAC address sync state, default True
check_ip_addresses	No	Check IP address sync state, default True
check_bgp_peerings	No	Check BGP peering sync state, default True
Nornir filters	No	Host filters such as FL, FB, FG, FC, or FN

At least one explicit device or Nornir host filter must resolve to a device.

Output

{
    "result": {
        "ceos-spine-1": {
            "in_sync": False,
            "interfaces": True,
            "mac_addresses": False,
            "ip_addresses": True,
            "bgp_peerings": True,
        },
    },
    "diff": {
        "interfaces": {},
        "mac_addresses": {},
        "ip_addresses": {},
        "bgp_peerings": {},
    },
}

A category is considered in sync when the corresponding dry-run reports no pending creates, updates, or deletes.

Notes / Gotchas

• No data is written to NetBox.
• Result.diff contains the raw dry-run detail from each enabled sub-task.
• Device names can be supplied directly or resolved from Nornir filters.

Examples

CLIPython

Check all sync categories for explicit devices:

nf#netbox check-sync devices devices ceos-leaf-1 ceos-leaf-2

Check only interface and IP address sync:

nf#netbox check-sync devices devices ceos-leaf-1 check-mac-addresses false check-bgp-peerings false

Resolve devices using a Nornir group filter:

nf#netbox check-sync devices FG leafs

Check against a NetBox branch:

nf#netbox check-sync devices devices ceos-leaf-1 branch my-branch

Context manager:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1", "ceos-leaf-2"],
        },
    )

Direct lifecycle:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1"],
            "check_mac_addresses": False,
            "check_bgp_peerings": False,
        },
    )

    filtered_result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "FG": "leafs",
        },
    )
finally:
    nf.destroy()

NORFAB Netbox Check Device Sync Command Shell Reference

NorFab shell supports these command options for Netbox check_device_sync task:

nf# man tree netbox.check-sync.devices
root
└── netbox:    Netbox service
    └── check-sync:    Check if Netbox data is in sync with live device state
        └── devices:    Check if device data in NetBox is in sync with live device state
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── branch:    Branching plugin branch name to use
            ├── devices:    List of NetBox devices to check sync state for
            ├── check-interfaces:    Check interfaces sync state, default 'True'
            ├── check-mac-addresses:    Check MAC addresses sync state, default 'True'
            ├── check-ip-addresses:    Check IP addresses sync state, default 'True'
            ├── check-bgp-peerings:    Check BGP peerings sync state, default 'True'
            ├── FO:    Filter hosts using Filter Object
            ├── FB:    Filter hosts by name using Glob Patterns
            ├── FH:    Filter hosts by hostname
            ├── FC:    Filter hosts containment of pattern in name
            ├── FR:    Filter hosts by name using Regular Expressions
            ├── FG:    Filter hosts by group
            ├── FP:    Filter hosts by hostname using IP Prefix
            ├── FL:    Filter hosts by names list
            ├── FX:    Filter hosts excluding them by name
            └── FN:    Negate the match
nf#

Python API Reference

Check if NetBox device data is in sync with live device data.

Calls sync_device_interfaces, sync_mac_addresses, sync_device_ip, and sync_bgp_peerings in dry-run mode and produces a per-device report indicating which items are in sync and which are not.

Result.diff contains the full dry-run detail from each sub-task, keyed by sub-task name (interfaces, mac_addresses, ip_addresses, bgp_peerings).

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object.	required
instance	str	NetBox instance name.	None
timeout	int	Timeout in seconds for Nornir jobs. Defaults to 60.	600
devices	list	List of device names to check.	None
branch	str	NetBox branching plugin branch name.	None
check_interfaces	bool	Check interface sync state. Defaults to True.	True
check_mac_addresses	bool	Check MAC address sync state. Defaults to True.	True
check_ip_addresses	bool	Check IP address sync state. Defaults to True.	True
check_bgp_peerings	bool	Check BGP peering sync state. Defaults to True.	True
**kwargs	Any	Nornir host filter arguments (e.g. FL, FC, FB).	{}

Returns:

Name	Type	Description
Result	Result	Per-device sync summary keyed by device name:: { "": { "in_sync": True | False, "interfaces": True | False, "mac_addresses": True | False, "ip_addresses": True | False, "bgp_peerings": True | False, } }

Source code in norfab\workers\netbox_worker\devices_tasks.py

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=CheckDeviceSyncInput,
    output=CheckDeviceSyncResult,
    mcp={
        "annotations": {
            "title": "Check Device Sync",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def check_device_sync(
    self,
    job: Job,
    instance: Union[None, str] = None,
    timeout: int = 600,
    devices: Union[None, list] = None,
    branch: str = None,
    check_interfaces: bool = True,
    check_mac_addresses: bool = True,
    check_ip_addresses: bool = True,
    check_bgp_peerings: bool = True,
    **kwargs: Any,
) -> Result:
    """
    Check if NetBox device data is in sync with live device data.

    Calls ``sync_device_interfaces``, ``sync_mac_addresses``, ``sync_device_ip``,
    and ``sync_bgp_peerings`` in dry-run mode and produces a per-device report
    indicating which items are in sync and which are not.

    ``Result.diff`` contains the full dry-run detail from each sub-task, keyed by
    sub-task name (``interfaces``, ``mac_addresses``, ``ip_addresses``,
    ``bgp_peerings``).

    Args:
        job: NorFab Job object.
        instance (str, optional): NetBox instance name.
        timeout (int): Timeout in seconds for Nornir jobs. Defaults to 60.
        devices (list, optional): List of device names to check.
        branch (str, optional): NetBox branching plugin branch name.
        check_interfaces (bool): Check interface sync state. Defaults to True.
        check_mac_addresses (bool): Check MAC address sync state. Defaults to True.
        check_ip_addresses (bool): Check IP address sync state. Defaults to True.
        check_bgp_peerings (bool): Check BGP peering sync state. Defaults to True.
        **kwargs: Nornir host filter arguments (e.g. ``FL``, ``FC``, ``FB``).

    Returns:
        Result: Per-device sync summary keyed by device name::

            {
                "<device>": {
                    "in_sync": True | False,
                    "interfaces":    True | False,
                    "mac_addresses": True | False,
                    "ip_addresses":  True | False,
                    "bgp_peerings":  True | False,
                }
            }
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:check_device_sync",
        result={},
        resources=[instance],
        diff={},
    )

    # resolve devices from Nornir filters
    if kwargs:
        job.event("resolving devices from Nornir filters")
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)
        job.event(
            f"resolved {len(nornir_hosts)} device(s) from Nornir filters, "
            f"{len(devices)} total device(s) selected"
        )

    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    log.info(
        f"{self.name} - Check device sync for {len(devices)} device(s) in '{instance}'"
    )
    job.event(f"checking sync state for {len(devices)} device(s)")

    # initialize per-device result structure
    for device in devices:
        ret.result[device] = {}

    # --- check interfaces ---
    if check_interfaces:
        job.event("checking interfaces sync state")
        intf_result = self.sync_device_interfaces(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if intf_result.errors:
            ret.errors.extend(intf_result.errors)
        for device, data in intf_result.result.items():
            in_sync = (
                not data.get("create")
                and not data.get("update")
                and not data.get("delete")
            )
            ret.result.setdefault(device, {})["interfaces"] = in_sync
        ret.diff["interfaces"] = intf_result.result

    # --- check MAC addresses ---
    if check_mac_addresses:
        job.event("checking MAC addresses sync state")
        mac_result = self.sync_mac_addresses(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if mac_result.errors:
            ret.errors.extend(mac_result.errors)
        for device, data in mac_result.result.items():
            in_sync = not data.get("created") and not data.get("updated")
            ret.result.setdefault(device, {})["mac_addresses"] = in_sync
        ret.diff["mac_addresses"] = mac_result.result

    # --- check IP addresses ---
    if check_ip_addresses:
        job.event("checking IP addresses sync state")
        ip_result = self.sync_device_ip(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if ip_result.errors:
            ret.errors.extend(ip_result.errors)
        for device, data in ip_result.result.items():
            in_sync = not data.get("created") and not data.get("updated")
            ret.result.setdefault(device, {})["ip_addresses"] = in_sync
        ret.diff["ip_addresses"] = ip_result.result

    # --- check BGP peerings ---
    if check_bgp_peerings:
        job.event("checking BGP peerings sync state")
        bgp_result = self.sync_bgp_peerings(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if bgp_result.errors:
            ret.errors.extend(bgp_result.errors)
        for device, data in bgp_result.result.items():
            in_sync = (
                not data.get("create")
                and not data.get("update")
                and not data.get("delete")
            )
            ret.result.setdefault(device, {})["bgp_peerings"] = in_sync
        ret.diff["bgp_peerings"] = bgp_result.result

    checked_categories = {
        "interfaces": check_interfaces,
        "mac_addresses": check_mac_addresses,
        "ip_addresses": check_ip_addresses,
        "bgp_peerings": check_bgp_peerings,
    }
    for device, device_data in ret.result.items():
        device_data["in_sync"] = all(
            device_data.get(category) is True
            for category, checked in checked_categories.items()
            if checked
        )

    log.info(
        f"{self.name} - Check device sync complete for {len(ret.result)} device(s)"
    )
    return ret