Netbox Get BGP Peerings Task

task api name: get_bgp_peerings

This task integrates with Netbox BGP Plugin and allows to fetch devices' BGP peerings.

How It Works

• Requires NetBox BGP plugin: The worker verifies the plugin is installed on the target instance before proceeding.
• Resolves device IDs: It calls get_devices() to map provided device names to NetBox device IDs for accurate API queries.
• Fetches sessions via REST: Uses pynetbox plugins.bgp.session.filter(device_id=...) to retrieve sessions per device.
• Returns structured data: The result is a dictionary keyed by device name; each device contains a dictionary keyed by BGP session name, with the full session dict as value.
• Smart caching: Per-device cache key get_bgp_peerings::<device> is used. Modes:
• True: Uses cache when up-to-date; performs smart update by comparing last_updated and fetching only changed/new sessions.
• False: Bypasses cache entirely and does not write to cache.
• refresh: Forces re-fetch from NetBox and overwrites cache.
• force: Returns cached data if present without freshness checks.

Sample BGP session data retrieved from Netbox:

"fceos4": {
    "fceos4-fceos5-eth105": {
        "comments": "",
        "created": "2025-12-31T08:17:39.168208Z",
        "custom_fields": {},
        "description": "BGP peering between fceos4 and fceos5 on eth105",
        "device": {
            "description": "",
            "display": "fceos4 (UUID-123451)",
            "id": 111,
            "name": "fceos4",
            "url": "http://192.168.1.210:8000/api/dcim/devices/111/"
        },
        "display": "fceos4 (UUID-123451):fceos4-fceos5-eth105",
        "export_policies": [],
        "id": 4,
        "import_policies": [],
        "last_updated": "2025-12-31T08:17:39.168231Z",
        "local_address": {
            "address": "10.0.2.1/30",
            "description": "",
            "display": "10.0.2.1/30",
            "family": {
                "label": "IPv4",
                "value": 4
            },
            "id": 123,
            "url": "http://192.168.1.210:8000/api/ipam/ip-addresses/123/"
        },
        "local_as": {
            "asn": 65100,
            "description": "BGP ASN for fceos4",
            "display": "AS65100",
            "id": 3,
            "url": "http://192.168.1.210:8000/api/ipam/asns/3/"
        },
        "name": "fceos4-fceos5-eth105",
        "peer_group": {
            "description": "Test BGP peer group 1 for standard peerings",
            "display": "TEST_BGP_PEER_GROUP_1",
            "id": 9,
            "name": "TEST_BGP_PEER_GROUP_1",
            "url": "http://192.168.1.210:8000/api/plugins/bgp/bgppeergroup/9/"
        },
        "prefix_list_in": null,
        "prefix_list_out": null,
        "remote_address": {
            "address": "10.0.2.2/30",
            "description": "",
            "display": "10.0.2.2/30",
            "family": {
                "label": "IPv4",
                "value": 4
            },
            "id": 124,
            "url": "http://192.168.1.210:8000/api/ipam/ip-addresses/124/"
        },
        "remote_as": {
            "asn": 65101,
            "description": "BGP ASN for fceos5",
            "display": "AS65101",
            "id": 4,
            "url": "http://192.168.1.210:8000/api/ipam/asns/4/"
        },
        "site": {
            "description": "",
            "display": "SALTNORNIR-LAB",
            "id": 16,
            "name": "SALTNORNIR-LAB",
            "slug": "saltnornir-lab",
            "url": "http://192.168.1.210:8000/api/dcim/sites/16/"
        },
        "status": {
            "label": "Active",
            "value": "active"
        },
        "tags": [],
        "tenant": {
            "description": "",
            "display": "SALTNORNIR",
            "id": 11,
            "name": "SALTNORNIR",
            "slug": "saltnornir",
            "url": "http://192.168.1.210:8000/api/tenancy/tenants/11/"
        },
        "url": "http://192.168.1.210:8000/api/plugins/bgp/bgpsession/4/",
        "virtualmachine": null
    },
    "fceos4-fceos5-eth106": {

        ...etc...

Gotchas

• Supported and tested Netbox version is 4.4 onwards.
• NetBox BGP plugin required: If missing, the task fails early with an error. Confirm plugin availability and version compatibility.
• Device name must exist: Unknown devices are skipped with warnings; verify names beforehand or use get_devices to inspect inventory.
• Session key uniqueness: Sessions in the result are keyed by name. If session names are not unique per device, later entries overwrite earlier ones.
• Partial-field queries: Smart update relies on fields="id,last_updated,name". Older Netbox versions may not support fields, impacting cache comparison.
• Large datasets: Fetching many devices or sessions may be slow; prefer cache or limit devices for interactive runs.

NORFAB Netbox Get BGP Peerings Command Shell Reference

NorFab shell supports these command options for Netbox get_bgp_peerings task:

nf#man tree netbox.get.bgp_peerings
root
└── netbox:    Netbox service
    └── get:    Query data from Netbox
        └── bgp-peerings:    Query Netbox BGP Peerings data
            ├── 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
            ├── dry-run:    Only return query content, do not run it
            ├── branch:    Branching plugin branch name to use
            ├── *devices:    Device names to query data for
            └── cache:    How to use cache, default 'True'
nf#

Python API Reference

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