Netbox Sync Device IP Task

task api name: sync_device_ip

The Netbox Sync Device IP Task synchronizes IP address assignments from live network devices into NetBox using a normalized desired/current state model and bulk-reconciliation. The task computes an explicit action plan — create, update, or mark in-sync — and applies it in a single pair of bulk API calls.

How It Works

The task follows a three-step pipeline:

1. Collect live state — Run a Nornir parse_ttp job against the target devices to collect live IPv4 and IPv6 addresses per interface.
2. Fetch NetBox state — Retrieve all existing IP address objects from NetBox that match any discovered address value.
3. Reconcile — Compare live addresses against NetBox records and classify each IP as:
   • create — address not in NetBox at all; a new record is created and assigned to the interface
   • update — address exists in NetBox but is unassigned or has a stale role/VRF; the record is updated
   • in_sync — address already assigned to the correct interface with the correct role and VRF; no change needed

Roles are assigned automatically:

• Interfaces whose name starts with loopback or lo → role loopback
• Addresses that fall within any configured anycast_ranges prefix → role anycast
• All other addresses → no role (standard unicast)

1. Client submits an on-demand request to the NorFab Netbox worker to sync device IP addresses
2. Netbox worker sends a job request to the Nornir service to fetch live interface data from devices
3. Nornir service collects IP address data from the network using parse_ttp
4. Nornir returns normalized interface and IP data to the Netbox worker
5. Netbox worker reconciles live addresses against NetBox and applies bulk create/update operations

Result Structure

Both dry-run and live-run modes return the same per-device structure keyed by device name:

{
    "<device>": {
        "created": ["10.0.0.1/31", "192.168.1.1/32"],
        "updated": ["10.0.0.3/31"],
        "in_sync": ["10.0.0.5/31", "2001:db8::1/128"]
    }
}

Dry-run mode (dry_run=True) returns what would be created or updated without making any changes to NetBox.

Live-run mode (dry_run=False, default) applies changes and returns the same structure showing what was done.

Filtering

IP addresses and interfaces can be scoped before reconciliation. All filters are applied at the live-data collection phase, so excluded addresses are completely ignored — they are neither created nor updated.

• filter_by_name — glob pattern matched against interface names, e.g. "Loopback*" or "Ethernet[1-4]". Interfaces not matching are skipped entirely.
• filter_by_description — glob pattern matched against interface descriptions, e.g. "uplink*". Interfaces not matching are skipped.
• filter_by_prefix — CIDR prefix such as "10.0.0.0/8". Only IP addresses whose host address falls within the prefix are included; supports both IPv4 and IPv6.
• filter_by_ip — glob pattern matched against the host portion of the IP address (without prefix length), e.g. "10.0.1.*". Only matching addresses are included.

Multiple filters combine as intersection — all specified conditions must be satisfied for an IP to be included.

Anycast Support

IP addresses in one or more anycast_ranges prefixes are assigned the anycast role. NetBox allows multiple IP address records with the same value when the role is anycast, so each device gets its own record. Without anycast_ranges, a second device trying to use the same IP address triggers a duplicate-conflict error.

Set anycast_ranges to a prefix string or list of prefixes:

anycast_ranges = "10.0.250.0/24"
anycast_ranges = ["10.0.250.0/24", "2001:db8:ffff::/48"]

Process Prefixes

When create_prefixes=True the task also creates a NetBox prefix record for the network of each discovered IP address (e.g. 10.0.1.0/31 for 10.0.1.1/31). Existing prefixes are never updated or deleted — this is a create-only, idempotent operation. Site and VRF are propagated from the device and interface context.

Branching Support

The task is branch-aware and can push changes into a NetBox branch. The Netbox Branching Plugin must be installed. Specify the branch parameter; the branch is created automatically if it does not already exist.

Duplicate IP Guard

Before executing bulk writes the task checks that no non-anycast IP address appears more than once across the combined create and update payloads. If a duplicate is detected, all copies are removed from the payload and an error is recorded in the result, preventing NetBox from receiving conflicting assignments in a single request.

Examples

CLIPython

Sync IP addresses for a list of devices:

nf#netbox sync ip-addresses devices ceos-spine-1 ceos-spine-2

Preview changes without writing to NetBox (dry run):

nf#netbox sync ip-addresses devices ceos-spine-1 dry-run

Sync only loopback interface addresses:

nf#netbox sync ip-addresses devices ceos-spine-1 filter-by-name "Loopback*"

Sync only addresses within a specific prefix:

nf#netbox sync ip-addresses devices ceos-spine-1 filter-by-prefix "10.3.0.0/16"

Classify addresses in an anycast range and sync all devices:

nf#netbox sync ip-addresses devices ceos-spine-1 ceos-spine-2 ceos-leaf-1 anycast-ranges 10.0.250.0/24

Also create prefix records for each discovered IP subnet:

nf#netbox sync ip-addresses devices ceos-spine-1 create-prefixes

Sync into a NetBox branch:

nf#netbox sync ip-addresses devices ceos-spine-1 ceos-spine-2 branch sprint-42-ips

Sync using Nornir host filters instead of explicit device names:

nf#netbox sync ip-addresses FC spine

from norfab.core.nfapi import NorFab

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

# sync IP addresses for specific devices
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
    },
)

# dry run — preview creates/updates without writing to NetBox
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "dry_run": True,
    },
)

# restrict sync to loopback interfaces only
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_name": "Loopback*",
    },
)

# restrict sync to addresses within a specific prefix
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "filter_by_prefix": "10.3.0.0/16",
    },
)

# restrict sync to addresses matching a glob pattern
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_ip": "10.3.4.*",
    },
)

# restrict sync to interfaces with a specific description pattern
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_description": "uplink*",
    },
)

# classify addresses in anycast ranges and sync all fabric devices
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2", "ceos-leaf-1", "ceos-leaf-2"],
        "anycast_ranges": "10.0.250.0/24",
    },
)

# multiple anycast ranges (IPv4 and IPv6)
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "anycast_ranges": ["10.0.250.0/24", "2001:db8:ffff::/48"],
    },
)

# also create prefix records for each discovered IP subnet
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "create_prefixes": True,
    },
)

# sync into a NetBox branch
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "branch": "sprint-42-ips",
    },
)

# use Nornir host filters instead of explicit device names
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "FC": "spine",
    },
)

# combine filters: loopback interfaces within a specific prefix, dry run
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "filter_by_name": "Loopback*",
        "filter_by_prefix": "10.3.4.0/24",
        "dry_run": True,
    },
)

nf.destroy()

NORFAB Netbox Sync Device IP Command Shell Reference

NorFab shell supports these command options for the sync_device_ip task:

nf# man tree netbox.sync.ip-addresses
root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── ip-addresses:    Sync device IP addresses with NetBox
            ├── timeout:    Job timeout in seconds
            ├── 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
            ├── dry-run:    Return reconciliation plan without pushing changes to NetBox
            ├── devices:    List of NetBox device names to sync
            ├── anycast-ranges:    IP prefix(es) used to classify addresses as anycast role
            ├── create-prefixes:    Create missing IP prefix records for each discovered address
            ├── filter-by-name:    Glob pattern to restrict sync by interface name, e.g. 'Loopback*'
            ├── filter-by-description:    Glob pattern to restrict sync by interface description
            ├── filter-by-prefix:    CIDR prefix to restrict sync to addresses within it, e.g. '10.0.0.0/8'
            ├── filter-by-ip:    Glob pattern to restrict sync by IP host address, e.g. '10.0.1.*'
            ├── branch:    Branching plugin branch name to push changes into
            ├── FO:    Filter Nornir hosts using Filter Object
            ├── FB:    Filter Nornir hosts by name using Glob Patterns
            ├── FH:    Filter Nornir hosts by hostname
            ├── FC:    Filter Nornir hosts by name containment
            ├── FR:    Filter Nornir hosts by name using Regular Expressions
            ├── FG:    Filter Nornir hosts by group
            ├── FP:    Filter Nornir hosts by hostname using IP Prefix
            ├── FL:    Filter Nornir hosts by names list
            ├── FM:    Filter Nornir hosts by platform
            └── FN:    Negate the Nornir host filter match
nf#

Python API Reference

Synchronize IP addresses from live devices into NetBox.

The task follows a three-step pipeline:

1. Collect live state: Run a Nornir parse_ttp get interfaces job against devices to collect live IP addresses per interface.
2. Fetch NetBox state: Retrieve existing IP address objects from NetBox.
3. Reconcile: Create new IP address objects, update existing unassigned or stale ones, and mark already-correct entries as in-sync.

Dry-run mode (dry_run=True): returns the reconciliation plan without making any changes. Result is keyed by device name::

{
    "<device>": {
        "created": ["10.0.0.1/31", ...],
        "updated": ["10.0.0.3/31", ...],
        "in_sync": ["10.0.0.5/31", ...]
    }
}

Live-run mode (dry_run=False, default): applies changes and returns the same structure showing what was done.

Side Effects

1. When overlapping IP discovered in Netbox and it is part of anycast range, existing IP role update to anycast

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
timeout	int	Timeout in seconds for the Nornir parse_ttp job.	60
devices	list	List of device names to sync.	None
branch	str	NetBox branch name to use.	None
anycast_ranges	list	IP prefix(es) used to classify IP addresses as anycast role, e.g. '10.3.250.0/24'.	None
ignore_ranges	list	Prefixes to ignore IP addresses for, includes by default 127.0.0.0/8, 224.0.0.0/24 and others	None
create_prefixes	bool	If True, create missing IP prefixes in NetBox for each discovered IP address. No updates or deletions are done.	False
filter_by_name	str	Glob pattern to restrict which interfaces are included by name, e.g. 'Loopback*' or 'Eth*'.	None
filter_by_description	str	Glob pattern to restrict which interfaces are included by description, e.g. 'uplink*'.	None
filter_by_prefix	str	IP prefix to restrict which IP addresses are included, e.g. '10.0.0.0/8'. Skips IPs not within this prefix.	None
filter_by_ip	str	Glob pattern to restrict which IP addresses are included, e.g. '10.0.*'.	None
**kwargs	Any	Additional Nornir host filter keyword arguments passed to parse_ttp (e.g. FL, FC, FB).	{}

Returns:

Name	Type	Description
Result	Result	Per-device action summary with created, updated, and in_sync IP address lists.

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

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncDeviceIpInput,
)
def sync_device_ip(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    timeout: int = 60,
    devices: Union[None, list] = None,
    branch: str = None,
    anycast_ranges: Union[None, list] = None,
    ignore_ranges: Union[None, list] = None,
    create_prefixes: bool = False,
    filter_by_name: Union[None, str] = None,
    filter_by_description: Union[None, str] = None,
    filter_by_prefix: Union[None, str] = None,
    filter_by_ip: Union[None, str] = None,
    **kwargs: Any,
) -> Result:
    """
    Synchronize IP addresses from live devices into NetBox.

    The task follows a three-step pipeline:

    1. **Collect live state**: Run a Nornir ``parse_ttp`` get interfaces job against
       devices to collect live IP addresses per interface.
    2. **Fetch NetBox state**: Retrieve existing IP address objects from NetBox.
    3. **Reconcile**: Create new IP address objects, update existing unassigned or
       stale ones, and mark already-correct entries as in-sync.

    **Dry-run mode** (``dry_run=True``): returns the reconciliation plan without
    making any changes. Result is keyed by device name::

    ```
    {
        "<device>": {
            "created": ["10.0.0.1/31", ...],
            "updated": ["10.0.0.3/31", ...],
            "in_sync": ["10.0.0.5/31", ...]
        }
    }
    ```

    **Live-run mode** (``dry_run=False``, default): applies changes and returns
    the same structure showing what was done.

    **Side Effects**

    1. When overlapping IP discovered in Netbox and it is part of anycast range,
        existing IP role update to `anycast`

    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.
        timeout (int, optional): Timeout in seconds for the Nornir parse_ttp job.
        devices (list, optional): List of device names to sync.
        branch (str, optional): NetBox branch name to use.
        anycast_ranges (list, optional): IP prefix(es) used to classify
            IP addresses as anycast role, e.g. ``'10.3.250.0/24'``.
        ignore_ranges (list, optional): Prefixes to ignore IP addresses for, includes
            by default 127.0.0.0/8, 224.0.0.0/24 and others
        create_prefixes (bool, optional): If True, create missing IP prefixes in
            NetBox for each discovered IP address. No updates or deletions are done.
        filter_by_name (str, optional): Glob pattern to restrict which interfaces
            are included by name, e.g. ``'Loopback*'`` or ``'Eth*'``.
        filter_by_description (str, optional): Glob pattern to restrict which
            interfaces are included by description, e.g. ``'uplink*'``.
        filter_by_prefix (str, optional): IP prefix to restrict which IP addresses
            are included, e.g. ``'10.0.0.0/8'``. Skips IPs not within this prefix.
        filter_by_ip (str, optional): Glob pattern to restrict which IP addresses
            are included, e.g. ``'10.0.*'``.
        **kwargs: Additional Nornir host filter keyword arguments passed to
            ``parse_ttp`` (e.g. ``FL``, ``FC``, ``FB``).

    Returns:
        Result: Per-device action summary with ``created``, ``updated``, and
            ``in_sync`` IP address lists.
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_ip",
        result={},
        resources=[instance],
        dry_run=dry_run,
    )
    nb = self._get_pynetbox(instance, branch=branch)
    log.info(
        f"{self.name} - Sync device IP: Processing {len(devices)} device(s) in '{instance}'"
    )

    # pre-process ranges
    ignore_ranges = ignore_ranges or [
        "127.0.0.0/8",
        "224.0.0.0/24",
        "fe80::/10",
        "ff02::/16",
    ]
    if isinstance(anycast_ranges, str):
        anycast_ranges = [anycast_ranges]
    if isinstance(ignore_ranges, str):
        ignore_ranges = [ignore_ranges]
    # convert prefix ranges to IP address objects
    anycast_nets = [
        ipaddress.ip_network(str(pfx), strict=False) for pfx in anycast_ranges or []
    ]
    ignore_nets = [
        ipaddress.ip_network(str(pfx), strict=False) for pfx in ignore_ranges or []
    ]
    filter_prefix_net = (
        ipaddress.ip_network(filter_by_prefix, strict=False)
        if filter_by_prefix
        else None
    )

    # source additional hosts from Nornir filters
    if kwargs:
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)

    if not devices:
        ret.errors.append("no devices specified")
        ret.failed = True
        return ret

    job.event(f"syncing IP addresses for {len(devices)} devices")

    # filter out devices not defined in NetBox
    nb_devices_data = {
        d.name: {
            "id": d.id,
            "name": d.name,
            "site_id": d.site.id if d.site else None,
        }
        for d in self.bulk_filter(
            nb.dcim.devices, "name", devices, fields="id,name,site"
        )
    }
    for d in list(devices):
        if d not in nb_devices_data:
            msg = f"{d} - device not found in Netbox"
            log.error(msg)
            job.event(msg, severity="ERROR")
            ret.errors.append(msg)
            devices.remove(d)
    if not devices:
        ret.failed = True
        return ret

    # gather live interface data from Nornir parse_ttp
    job.event(f"retrieving live interfaces for {len(devices)} devices")
    parse_data = self.client.run_job(
        "nornir",
        "parse_ttp",
        kwargs={"get": "interfaces", "FL": devices},
        workers="all",
        timeout=timeout,
    )

    # normalise live data from Nornir parse_ttp results into {device: {intf: intf_data}}
    # applying interface name and description filters at collection phase
    normalised_live_all = {}
    for wname, wdata in parse_data.items():
        if wdata.get("failed"):
            log.warning(f"{wname} - failed to parse devices")
            continue
        for device_name, host_interfaces in wdata["result"].items():
            filtered = {}
            for intf in host_interfaces:
                intf_name = intf["name"]
                intf_description = intf.get("description") or ""
                if filter_by_name and not fnmatch.fnmatch(
                    intf_name, filter_by_name
                ):
                    continue
                if filter_by_description and not fnmatch.fnmatch(
                    intf_description, filter_by_description
                ):
                    continue
                # apply IP-level filters if needed
                if filter_prefix_net or filter_by_ip or ignore_nets:
                    filtered_ipv4 = []
                    filtered_ipv6 = []
                    for ip in intf.get("ipv4_addresses") or []:
                        ip_addr = ipaddress.ip_interface(str(ip)).ip
                        if filter_prefix_net and ip_addr not in filter_prefix_net:
                            continue
                        if filter_by_ip and not fnmatch.fnmatch(
                            str(ip_addr), filter_by_ip
                        ):
                            continue
                        if ignore_nets and any(
                            ip_addr in net for net in ignore_nets
                        ):
                            continue
                        filtered_ipv4.append(ip)
                    for ip in intf.get("ipv6_addresses") or []:
                        ip_addr = ipaddress.ip_interface(str(ip)).ip
                        if filter_prefix_net and ip_addr not in filter_prefix_net:
                            continue
                        if filter_by_ip and not fnmatch.fnmatch(
                            str(ip_addr), filter_by_ip
                        ):
                            continue
                        if ignore_nets and any(
                            ip_addr in net for net in ignore_nets
                        ):
                            continue
                        filtered_ipv6.append(ip)
                    # skip interface entirely if all IPs were filtered out
                    if not filtered_ipv4 and not filtered_ipv6:
                        continue
                    intf = {
                        **intf,
                        "ipv4_addresses": filtered_ipv4,
                        "ipv6_addresses": filtered_ipv6,
                    }
                filtered[intf_name] = intf
            normalised_live_all[device_name] = filtered

    # fetch interfaces data from NetBox to resolve interface IDs
    nb_interfaces_result = self.get_interfaces(
        job=job,
        instance=instance,
        branch=branch,
        devices=devices,
        ip_addresses=False,
        cache="refresh",
    )
    if nb_interfaces_result.errors:
        ret.errors.extend(nb_interfaces_result.errors)
        ret.failed = True
        return ret

    # per-device result tracking
    device_results = {
        device_name: {
            "created": [],
            "updated": [],
            "in_sync": [],
        }
        for device_name in devices
    }
    ret.result = device_results

    # collect all discovered IP addresses and prefixes
    all_ip_live = []
    all_prefixes_live = []
    for device_name, interfaces in normalised_live_all.items():
        if device_name not in nb_devices_data:
            continue
        nb_device = nb_devices_data[device_name]
        nb_raw = nb_interfaces_result.result.get(device_name, {})
        for intf_name, intf_data in interfaces.items():
            if intf_name not in nb_raw:
                continue
            for ip in (intf_data.get("ipv4_addresses") or []) + (
                intf_data.get("ipv6_addresses") or []
            ):
                vrf = resolve_vrf(intf_data["vrf"], nb, job, ret, self.name)
                all_ip_live.append(
                    {
                        "device": device_name,
                        "interface": intf_name,
                        "address": ip,
                        "vrf": vrf,
                        "role": resolve_ip_role(ip, intf_name, anycast_nets),
                        "assigned_object_type": "dcim.interface",
                        "assigned_object_id": nb_raw[intf_name]["id"],
                    }
                )
                if create_prefixes:
                    all_prefixes_live.append(
                        {
                            "prefix": make_prefix_from_ip(ip),
                            "vrf": vrf,
                            "site": nb_device["site_id"],
                        }
                    )

    if not all_ip_live:
        log.info(
            f"{self.name} - Sync device IP: no IP addresses found in live data"
        )
        return ret

    # fetch existing IP addresses from NetBox
    nb_ips = [
        {
            "id": ip.id,
            "address": ip.address,
            "vrf": ip.vrf.id if ip.vrf else None,
            "role": str(ip.role).lower(),
            "assigned_object_id": (
                ip.assigned_object.id if ip.assigned_object else None
            ),
            "device": (
                ip.assigned_object.device.name if ip.assigned_object else None
            ),
            "interface": ip.assigned_object.name if ip.assigned_object else None,
        }
        for ip in self.bulk_filter(
            endpoint=nb.ipam.ip_addresses,
            filter_by_key="address",
            filter_by_values=[i["address"].split("/")[0] for i in all_ip_live],
            fields="id,address,vrf,role,assigned_object",
        )
    ]

    # process IP and Prefixes
    bulk_update_ip = {}  # {(device, intf, ip): {ip data}}
    bulk_create_ip = {}  # {(device, intf, ip): {ip data}}

    for ip_live in all_ip_live:
        device_name = ip_live.pop("device")
        intf_name = ip_live.pop("interface")
        key = (device_name, intf_name, ip_live["address"])
        # find existing NetBox IPs of same value
        ip_live_no_mask = ip_live["address"].split("/")[0]
        # do IP comparison ignoring mask, in case live and netbox IPs have mask mismatch
        matching_nb_ips = [
            i for i in nb_ips if i["address"].startswith(f"{ip_live_no_mask}/")
        ]
        # no existing IP found, create it
        if not matching_nb_ips:
            bulk_create_ip[key] = ip_live
            continue
        # check if existing IP already assigned to same interface
        for nb_ip in matching_nb_ips:
            if nb_ip["assigned_object_id"]:
                # ip already assigned to same interface
                if nb_ip["assigned_object_id"] == ip_live["assigned_object_id"]:
                    # check if vrf or role need an update
                    if any(
                        nb_ip[k] != ip_live[k]
                        for k in ["role", "vrf"]
                        if ip_live[k]
                    ):
                        ip_live["id"] = nb_ip["id"]
                        bulk_update_ip[key] = ip_live
                    else:
                        device_results[device_name]["in_sync"].append(
                            ip_live["address"]
                        )
                    break
        # no IP assigned to same interface
        else:
            for nb_ip in matching_nb_ips:
                # existing NB IP already assigned to an interface
                if nb_ip["assigned_object_id"]:
                    nb_ip_resolved_role = (
                        resolve_ip_role(
                            nb_ip["address"], nb_ip["interface"], anycast_nets
                        )
                        or nb_ip["role"]
                    )
                    # add existing IP to updates if its role needs correcting
                    if nb_ip["role"] != nb_ip_resolved_role:
                        nb_ip_key = (
                            nb_ip["device"],
                            nb_ip["interface"],
                            nb_ip["address"],
                        )
                        bulk_update_ip[nb_ip_key] = {
                            "id": nb_ip["id"],
                            "role": nb_ip_resolved_role,
                        }
                    # create anycast ip if existing and discovered IPs are anycast
                    if nb_ip_resolved_role == ip_live["role"] == "anycast":
                        bulk_create_ip[key] = ip_live
                        break
                    # existing ip role did not resolve to anycast - report conflict
                    if nb_ip_resolved_role != "anycast":
                        msg = (
                            f"duplicate non anycast ip found, {device_name}:{intf_name}->{ip_live['address']}, "
                            f"overlaps with {nb_ip['device']}:{nb_ip['interface']}->{nb_ip['address']}"
                        )
                        log.error(msg)
                        ret.errors.append(msg)
                        job.event(msg, severity="ERROR")
                        break
                # ip exists but not assigned to any interface - update it
                else:
                    ip_live["id"] = nb_ip["id"]
                    bulk_update_ip[key] = ip_live

    # make sure all devices present in results
    for key in list(bulk_create_ip) + list(bulk_update_ip):
        device_results.setdefault(
            key[0],
            {
                "created": [],
                "updated": [],
                "in_sync": [],
            },
        )

    if dry_run is True:
        for key in bulk_create_ip:
            device_name = key[0]
            device_results[device_name]["created"].append(key[2])
        for key in bulk_update_ip:
            device_name = key[0]
            device_results[device_name]["updated"].append(key[2])
        ret.dry_run = True
        return ret

    # check that update and create payloads have no non-anycast duplicate IPs
    # Netbox has a bug allowing to create duplicate IPs in single create request
    ip_address_seen = {}  # {address: [key, ...]}
    for key, ip_data in {**bulk_create_ip, **bulk_update_ip}.items():
        addr = key[2]
        role = ip_data.get("role") or ""
        if role != "anycast":
            ip_address_seen.setdefault(addr, []).append(key)
    for addr, dup_keys in ip_address_seen.items():
        if len(dup_keys) > 1:
            for key in dup_keys:
                device_name, intf_name, _ = key
                msg = (
                    f"found duplicate non-anycast IP {addr} in payload for "
                    f"{device_name}:{intf_name}, skipping"
                )
                log.error(msg)
                ret.errors.append(msg)
                job.event(msg, severity="ERROR")
                bulk_create_ip.pop(key, None)
                bulk_update_ip.pop(key, None)

    # update first, since existing IPs might change role to anycast
    if bulk_update_ip:
        try:
            nb.ipam.ip_addresses.update(list(bulk_update_ip.values()))
            job.event(f"updated {len(bulk_update_ip)} IP addresses")
            for key in bulk_update_ip:
                device_name = key[0]
                ip_address = key[2]
                device_results[device_name]["updated"].append(ip_address)
        except Exception as e:
            msg = f"failed to bulk update IP addresses: {e}"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")

    # create new IPs next
    if bulk_create_ip:
        try:
            nb.ipam.ip_addresses.create(list(bulk_create_ip.values()))
            job.event(f"created {len(bulk_create_ip)} IP addresses")
            for key in bulk_create_ip:
                device_name = key[0]
                ip_address = key[2]
                device_results[device_name]["created"].append(ip_address)
        except Exception as e:
            msg = f"failed to bulk create IP addresses: {e}"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")

    # process prefixes - create only, no updates or deletions
    if create_prefixes and all_prefixes_live:
        nb_prefixes = {
            (pfx.prefix, pfx.vrf.id if pfx.vrf else None)
            for pfx in nb.ipam.prefixes.filter(
                prefix=[p["prefix"] for p in all_prefixes_live if p["prefix"]],
                fields="id,prefix,vrf",
            )
        }
        bulk_create_prefixes = []
        seen_prefixes = set(nb_prefixes)
        for pfx_data in all_prefixes_live:
            if not pfx_data["prefix"]:
                continue
            pfx_key = (pfx_data["prefix"], pfx_data["vrf"])
            if pfx_key not in seen_prefixes:
                payload = {"prefix": pfx_data["prefix"]}
                if pfx_data["vrf"] is not None:
                    payload["vrf"] = pfx_data["vrf"]
                if pfx_data["site"] is not None:
                    payload["site"] = pfx_data["site"]
                bulk_create_prefixes.append(payload)
                seen_prefixes.add(pfx_key)
        if bulk_create_prefixes:
            try:
                nb.ipam.prefixes.create(bulk_create_prefixes)
                job.event(f"created {len(bulk_create_prefixes)} prefixes")
            except Exception as e:
                msg = f"failed to bulk create prefixes: {e}"
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")

    return ret