rename_stack_interfaces.py

#!/usr/bin/env python
#  All new devices will have interface naming copied from the
#  device-type template.  For virtual chassis members this will be
#  incorrect if they are not in the first position of the stack.  This
#  script will find and rename these interfaces.
from os import environ
from pprint import pprint

import pynetbox
from more_itertools import chunked
from rich.progress import track
from tenacity import retry
from tenacity import stop_after_attempt

url = "https://" + str(environ.get("NETBOX_HOST"))
nb = pynetbox.api(url, environ.get("NETBOX_TOKEN"))
part_numbers = ["WS-C3850-48F-E", "WS-C3850-48U-E"]


def main(chunk_size=500):
    interfaces = []
    for device in track(stack_members(), description="   Scanning devices"):
        interfaces.extend(modifications(device))

    # split the requests into chunks to avoid timeouts processing large batches
    for chunk in chunked(
        track(interfaces, description="Updating interfaces"), chunk_size
    ):
        rename_interfaces(chunk)


def stack_members():
    type_ids = [t.id for t in nb.dcim.device_types.filter(part_number=part_numbers)]
    return nb.dcim.devices.filter(device_type_id=type_ids, vc_position__n=1)


def modifications(device):
    match_prefix = "GigabitEthernet1/"
    new_prefix = match_prefix.replace("1", str(device.vc_position))
    return [
        {"id": iface.id, "name": iface.name.replace(match_prefix, new_prefix)}
        for iface in nb.dcim.interfaces.filter(
            device_id=device.id, name__ic=match_prefix
        )
    ]


# work around threading bug in netbox causing occasional 500 responses
@retry(stop=stop_after_attempt(3))
def rename_interfaces(interfaces):
    # use the API directly as pynetbox does not currently support bulk updates
    if interfaces:
        session = nb.http_session
        session.headers.update({"Authorization": f"Token {nb.token}"})
        r = session.patch(f"{url}/api/dcim/interfaces/", json=interfaces)


if __name__ == "__main__":
    main()