Netbox Get Connections Task

task api name: get_connections

Get Connections Sample Usage

NORFAB Netbox Get Connections Command Shell Reference

NorFab shell supports these command options for Netbox get_connections task:

Python API Reference

Retrieve interface connection details for specified devices from Netbox.

Parameters:

Name	Type	Description	Default
devices	list	List of device names to retrieve connections for.	required
instance	str	Instance name for the GraphQL query.	None
dry_run	bool	If True, perform a dry run without making actual changes.	False
cables	bool	if True includes interfaces' directly attached cables details	False

Returns:

Name	Type	Description
dict	dict	A dictionary containing connection details for each device.

Raises:

Type	Description
Exception	If there is an error in the GraphQL query or data retrieval process.

Source code in norfab\workers\netbox_worker.py

def get_connections(
    self,
    devices: list,
    instance: str = None,
    dry_run: bool = False,
    cables: bool = False,
) -> dict:
    """
    Retrieve interface connection details for specified devices from Netbox.

    Args:
        devices (list): List of device names to retrieve connections for.
        instance (str, optional): Instance name for the GraphQL query.
        dry_run (bool, optional): If True, perform a dry run without making actual changes.
        cables (bool, optional): if True includes interfaces' directly attached cables details

    Returns:
        dict: A dictionary containing connection details for each device.

    Raises:
        Exception: If there is an error in the GraphQL query or data retrieval process.
    """
    # form final result dictionary
    ret = Result(
        task=f"{self.name}:get_connections", result={d: {} for d in devices}
    )

    # form lists of fields to request from netbox
    cable_fields = """
        cable {
            type
            status
            tenant {name}
            label
            tags {name}
            length
            length_unit
            custom_fields
        }
    """
    if self.nb_version[0] == 4:
        interfaces_fields = [
            "name",
            "device {name}",
            """connected_endpoints {
            __typename 
            ... on InterfaceType {name device {name}}
            ... on ProviderNetworkType {name}
            }""",
        ]
    elif self.nb_version[0] == 3:
        interfaces_fields = [
            "name",
            "device {name}",
            """connected_endpoints {
            __typename 
            ... on InterfaceType {name device {name}}
            }""",
        ]
    interfaces_fields.append(
        """
        link_peers {
            __typename
            ... on InterfaceType {name device {name}}
            ... on FrontPortType {name device {name}}
            ... on RearPortType {name device {name}}
        }
    """
    )
    console_ports_fields = [
        "name",
        "device {name}",
        """connected_endpoints {
          __typename 
          ... on ConsoleServerPortType {name device {name}}
        }""",
        """link_peers {
          __typename
          ... on ConsoleServerPortType {name device {name}}
          ... on FrontPortType {name device {name}}
          ... on RearPortType {name device {name}}
        }""",
    ]
    console_server_ports_fields = [
        "name",
        "device {name}",
        """connected_endpoints {
          __typename 
          ... on ConsolePortType {name device {name}}
        }""",
        """link_peers {
          __typename
          ... on ConsolePortType {name device {name}}
          ... on FrontPortType {name device {name}}
          ... on RearPortType {name device {name}}
        }""",
    ]

    # check if need to include cables info
    if cables is True:
        interfaces_fields.append(cable_fields)
        console_ports_fields.append(cable_fields)
        console_server_ports_fields.append(cable_fields)

    # form query dictionary with aliases to get data from Netbox
    queries = {
        "interface": {
            "obj": "interface_list",
            "filters": {"device": devices},
            "fields": interfaces_fields,
        },
        "consoleport": {
            "obj": "console_port_list",
            "filters": {"device": devices},
            "fields": console_ports_fields,
        },
        "consoleserverport": {
            "obj": "console_server_port_list",
            "filters": {"device": devices},
            "fields": console_server_ports_fields,
        },
    }

    # retrieve full list of devices interface with all cables
    query_result = self.graphql(queries=queries, instance=instance, dry_run=dry_run)

    # return dry run result
    if dry_run:
        return query_result

    all_ports = query_result.result

    # extract interfaces
    for port_type, ports in all_ports.items():
        for port in ports:
            endpoints = port["connected_endpoints"]
            # skip ports that have no remote device connected
            if not endpoints or not all(i for i in endpoints):
                continue

            # extract required parameters
            cable = port.get("cable", {})
            device_name = port["device"]["name"]
            port_name = port["name"]
            link_peers = port["link_peers"]
            remote_termination_type = endpoints[0]["__typename"].lower()
            remote_termination_type = remote_termination_type.replace("type", "")

            # form initial connection dictionary
            connection = {
                "breakout": len(endpoints) > 1,
                "remote_termination_type": remote_termination_type,
                "termination_type": port_type,
            }

            # add remote connection details
            if remote_termination_type == "providernetwork":
                connection["remote_device"] = None
                connection["remote_interface"] = None
                connection["provider"] = endpoints[0]["name"]
            else:
                remote_interface = endpoints[0]["name"]
                if len(endpoints) > 1:
                    remote_interface = [i["name"] for i in endpoints]
                connection["remote_interface"] = remote_interface
                connection["remote_device"] = endpoints[0]["device"]["name"]

            # add cable and its peer details
            if cables:
                peer_termination_type = link_peers[0]["__typename"].lower()
                peer_termination_type = peer_termination_type.replace("type", "")
                cable["peer_termination_type"] = peer_termination_type
                cable["peer_device"] = link_peers[0].get("device", {}).get("name")
                cable["peer_interface"] = link_peers[0].get("name")
                if len(link_peers) > 1:  # handle breakout cable
                    cable["peer_interface"] = [i["name"] for i in link_peers]
                connection["cable"] = cable

            ret.result[device_name][port_name] = connection

    return ret