Helpers

Helper methods that are used across GSO.

available_interfaces_choices(router_id, speed)

Return a list of available interfaces for a given router and speed.

For Nokia routers, return a list of available interfaces. For Juniper routers, return a string.

Source code in gso/utils/helpers.py

def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
    """Return a list of available interfaces for a given router and speed.

    For Nokia routers, return a list of available interfaces.
    For Juniper routers, return a string.
    """
    if get_router_vendor(router_id) != Vendor.NOKIA:
        return None
    interfaces = {
        interface["name"]: f"{interface["name"]}  {interface["description"]}"
        for interface in NetboxClient().get_available_interfaces(router_id, speed)
    }
    return Choice("ae member", zip(interfaces.keys(), interfaces.items(), strict=True))  # type: ignore[arg-type]

available_interfaces_choices_including_current_members(router_id, speed, interfaces)

Return a list of available interfaces for a given router and speed including the current members.

For Nokia routers, return a list of available interfaces. For Juniper routers, return a string.

Source code in gso/utils/helpers.py

def available_interfaces_choices_including_current_members(
    router_id: UUID,
    speed: str,
    interfaces: list["IptrunkInterfaceBlock"],
) -> Choice | None:
    """Return a list of available interfaces for a given router and speed including the current members.

    For Nokia routers, return a list of available interfaces.
    For Juniper routers, return a string.
    """
    if get_router_vendor(router_id) != Vendor.NOKIA:
        return None

    available_interfaces = list(NetboxClient().get_available_interfaces(router_id, speed))
    available_interfaces.extend(
        [
            NetboxClient().get_interface_by_name_and_device(
                interface.interface_name,
                Router.from_subscription(router_id).router.router_fqdn,
            )
            for interface in interfaces
        ],
    )
    options = {
        interface["name"]: f"{interface["name"]}  {interface["description"]}" for interface in available_interfaces
    }
    return Choice("ae member", zip(options.keys(), options.items(), strict=True))  # type: ignore[arg-type]

available_lags_choices(router_id)

Return a list of available lags for a given router.

For Nokia routers, return a list of available lags. For Juniper routers, return None.

Source code in gso/utils/helpers.py

def available_lags_choices(router_id: UUID) -> Choice | None:
    """Return a list of available lags for a given router.

    For Nokia routers, return a list of available lags.
    For Juniper routers, return `None`.
    """
    if get_router_vendor(router_id) != Vendor.NOKIA:
        return None
    side_a_ae_iface_list = NetboxClient().get_available_lags(router_id)
    return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True))  # type: ignore[arg-type]

available_service_lags_choices(router_id)

Return a list of available lags for a given router for services.

For Nokia routers, return a list of available lags. For Juniper routers, return None.

Source code in gso/utils/helpers.py

def available_service_lags_choices(router_id: UUID) -> Choice | None:
    """Return a list of available lags for a given router for services.

    For Nokia routers, return a list of available lags.
    For Juniper routers, return `None`.
    """
    if get_router_vendor(router_id) != Vendor.NOKIA:
        return None
    side_a_ae_iface_list = NetboxClient().get_available_services_lags(router_id)
    return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True))  # type: ignore[arg-type]

get_router_vendor(router_id)

Retrieve the vendor of a router.

Parameters:

Name	Type	Description	Default
router_id	UUID	The UUID of the router.	required

Returns:

Type	Description
Vendor	The vendor of the router.

Source code in gso/utils/helpers.py

def get_router_vendor(router_id: UUID) -> Vendor:
    """Retrieve the vendor of a router.

    Args:
        router_id: The UUID of the router.

    Returns:
        The vendor of the router.
    """
    return Router.from_subscription(router_id).router.vendor

iso_from_ipv4(ipv4_address)

Calculate an ISO address, based on an IPv4 address.

Parameters:

Name	Type	Description	Default
ipv4_address	IPv4AddressType	The address that's to be converted	required

Returns:

Type	Description
str	An ISO-formatted address.

Source code in gso/utils/helpers.py

def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
    """Calculate an ISO address, based on an IPv4 address.

    Args:
        ipv4_address: The address that's to be converted

    Returns:
        An ISO-formatted address.
    """
    padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")]
    joined_octets = "".join(padded_octets)
    re_split = ".".join(re.findall("....", joined_octets))
    return f"49.51e5.0001.{re_split}.00"

generate_fqdn(hostname, site_name, country_code)

Generate an FQDN from a hostname, site name, and a country code.

Source code in gso/utils/helpers.py

def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
    """Generate an FQDN from a hostname, site name, and a country code."""
    oss = settings.load_oss_params()
    return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"

generate_lan_switch_interconnect_subnet_v4(site_internal_id)

Generate an IPv4 network in which a LAN Switch Interconnect resides, given a Site internal ID.

Source code in gso/utils/helpers.py

def generate_lan_switch_interconnect_subnet_v4(site_internal_id: int) -> IPv4NetworkType:
    """Generate an IPv4 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT

    result = str(ipam_oss.V4.containers[0]).split(".")[:2]  # Take the first two octets from the IPv4 network.
    result.append(str(site_internal_id))  # Append the side ID as the third octet.
    result.append(f"0/{ipam_oss.V4.mask}")  # Append the fourth octet, together with the netmask.

    return IPv4Network(".".join(result))

generate_lan_switch_interconnect_subnet_v6(site_internal_id)

Generate an IPv6 network in which a LAN Switch Interconnect resides, given a Site internal ID.

Source code in gso/utils/helpers.py

def generate_lan_switch_interconnect_subnet_v6(site_internal_id: int) -> IPv6NetworkType:
    """Generate an IPv6 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT

    result = IPv6Network(ipam_oss.V6.containers[0]).exploded[:17]  # Take the first 56 bits of the network
    result += str(hex(site_internal_id)[2:])  # Append the site internal id for bytes 57 to 64 as hexadecimal number
    result += f"::/{ipam_oss.V6.mask}"  # And fill the rest of the network with empty bits

    return IPv6Network(result)

generate_inventory_for_routers(router_role, exclude_routers=None, router_vendor=None, *, include_provisioning_routers=True)

Generate an Ansible-compatible inventory for executing playbooks.

Contains all active routers of a specific role. Optionally, routers can be excluded from the inventory.

Parameters:

Name	Type	Description	Default
router_role	RouterRole	The role of the routers to include in the inventory.	required
exclude_routers	list[str] | None	List of routers to exclude from the inventory.	None
router_vendor	Vendor | None	The vendor of the routers to include in the inventory.	None
include_provisioning_routers	bool	Include routers that are in a PROVISIONING state.	True

Returns:

Type	Description
dict	A dictionary representing the inventory of active routers.

Source code in gso/utils/helpers.py

def generate_inventory_for_routers(
    router_role: RouterRole,
    exclude_routers: list[str] | None = None,
    router_vendor: Vendor | None = None,
    *,
    include_provisioning_routers: bool = True,
) -> dict:
    """Generate an Ansible-compatible inventory for executing playbooks.

    Contains all active routers of a specific role. Optionally, routers can be excluded from the inventory.

    Args:
        router_role: The role of the routers to include in the inventory.
        exclude_routers: List of routers to exclude from the inventory.
        router_vendor: The vendor of the routers to include in the inventory.
        include_provisioning_routers: Include routers that are in a `PROVISIONING` state.

    Returns:
        A dictionary representing the inventory of active routers.
    """
    lifecycles = (
        [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
        if include_provisioning_routers
        else [SubscriptionLifecycle.ACTIVE]
    )
    all_routers = [
        Router.from_subscription(r["subscription_id"]) for r in get_router_subscriptions(lifecycles=lifecycles)
    ]
    exclude_routers = exclude_routers or []

    return {
        "all": {
            "hosts": {
                router.router.router_fqdn: {
                    "lo4": str(router.router.router_lo_ipv4_address),
                    "lo6": str(router.router.router_lo_ipv6_address),
                    "vendor": str(router.router.vendor),
                }
                for router in all_routers
                if router.router.router_role == router_role
                and router.router.router_fqdn not in exclude_routers
                and (router_vendor is None or router.router.vendor == router_vendor)
            }
        }
    }

calculate_recommended_minimum_links(iptrunk_number_of_members, iptrunk_speed)

Calculate the recommended minimum number of links for an IP trunk based on the number of members and speed.

If the IP trunk speed is 400G, the recommended minimum number of links is the number of members minus 1. Otherwise, the recommended minimum number of links is the number of members.

Parameters:

Name	Type	Description	Default
iptrunk_number_of_members	int	The number of members in the IP trunk.	required
iptrunk_speed	PhysicalPortCapacity	The speed of the IP trunk.	required

Returns:

Type	Description
int	The recommended minimum number of links for the IP trunk.

Source code in gso/utils/helpers.py

def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_speed: PhysicalPortCapacity) -> int:
    """Calculate the recommended minimum number of links for an IP trunk based on the number of members and speed.

    If the IP trunk speed is 400G, the recommended minimum number of links is the number of members minus 1.
    Otherwise, the recommended minimum number of links is the number of members.

    Args:
        iptrunk_number_of_members: The number of members in the IP trunk.
        iptrunk_speed: The speed of the IP trunk.

    Returns:
        The recommended minimum number of links for the IP trunk.
    """
    if iptrunk_speed == PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND:
        return iptrunk_number_of_members - 1
    return iptrunk_number_of_members

active_site_selector()

Generate a dropdown selector for choosing an active site in an input form.

Source code in gso/utils/helpers.py

def active_site_selector() -> Choice:
    """Generate a dropdown selector for choosing an active site in an input form."""
    site_subscriptions = {
        str(site["subscription_id"]): site["description"]
        for site in get_active_site_subscriptions(includes=["subscription_id", "description"])
    }

    return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True))  # type: ignore[arg-type]

active_router_selector()

Generate a dropdown selector for choosing an active Router in an input form.

Source code in gso/utils/helpers.py

def active_router_selector() -> Choice:
    """Generate a dropdown selector for choosing an active Router in an input form."""
    router_subscriptions = {
        str(router["subscription_id"]): router["description"]
        for router in get_active_router_subscriptions(includes=["subscription_id", "description"])
    }

    return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True))  # type: ignore[arg-type]

active_pe_router_selector(excludes=None)

Generate a dropdown selector for choosing an active PE Router in an input form.

Source code in gso/utils/helpers.py

def active_pe_router_selector(excludes: list[UUIDstr] | None = None) -> Choice:
    """Generate a dropdown selector for choosing an active PE Router in an input form."""
    excludes = excludes or []

    routers = {
        str(router.subscription_id): router.description
        for router in get_active_subscriptions_by_field_and_value("router_role", RouterRole.PE)
        if router.subscription_id not in excludes
    }

    return Choice("Select a router", zip(routers.keys(), routers.items(), strict=True))  # type: ignore[arg-type]

active_switch_selector()

Generate a dropdown selector for choosing an active Switch in an input form.

Source code in gso/utils/helpers.py

def active_switch_selector() -> Choice:
    """Generate a dropdown selector for choosing an active Switch in an input form."""
    switch_subscriptions = {
        str(switch["subscription_id"]): switch["description"]
        for switch in get_active_switch_subscriptions(includes=["subscription_id", "description"])
    }

    return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True))  # type: ignore[arg-type]

active_edge_port_selector(*, partner_id=None)

Generate a dropdown selector for choosing an active Edge Port in an input form.

Source code in gso/utils/helpers.py

def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> Choice:
    """Generate a dropdown selector for choosing an active Edge Port in an input form."""
    edge_ports = get_active_edge_port_subscriptions(partner_id=partner_id)

    options = {str(edge_port.subscription_id): edge_port.description for edge_port in edge_ports}

    return Choice(
        "Select an Edge Port",
        zip(options.keys(), options.items(), strict=True),  # type: ignore[arg-type]
    )

partner_choice()

Return a Choice object containing a list of available partners.

Source code in gso/utils/helpers.py

def partner_choice() -> Choice:
    """Return a Choice object containing a list of available partners."""
    partners = {partner.partner_id: partner.name for partner in get_all_partners()}

    return Choice("Select a partner", zip(partners.values(), partners.items(), strict=True))  # type: ignore[arg-type]

validate_edge_port_number_of_members_based_on_lacp(*, number_of_members, enable_lacp)

Validate the number of edge port members based on the LACP setting.

Parameters:

Name	Type	Description	Default
number_of_members	int	The number of members to validate.	required
enable_lacp	bool	Whether LACP is enabled or not.	required

Raises:

Type	Description
ValueError	If the number of members is greater than 1 and LACP is disabled.

Source code in gso/utils/helpers.py

def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int, enable_lacp: bool) -> None:
    """Validate the number of edge port members based on the LACP setting.

    Args:
        number_of_members: The number of members to validate.
        enable_lacp: Whether LACP is enabled or not.

    Raises:
        ValueError: If the number of members is greater than 1 and LACP is disabled.
    """
    if number_of_members > 1 and not enable_lacp:
        err_msg = "Number of members must be 1 if LACP is disabled."
        raise ValueError(err_msg)

generate_unique_vc_id(max_attempts=100)

Generate a unique 8-digit VC_ID starting with '11'.

This function attempts to generate an 8-digit VC_ID beginning with '11', checking its uniqueness before returning it. A maximum attempt limit is set to prevent infinite loops in case the ID space is saturated.

Parameters:

Name	Type	Description	Default
max_attempts	int	The maximum number of attempts to generate a unique ID.	100

Returns:

Type	Description
VC_ID | None	A unique VC_ID instance if successful, None if no unique ID is found.

Source code in gso/utils/helpers.py

def generate_unique_vc_id(max_attempts: int = 100) -> VC_ID | None:
    """Generate a unique 8-digit VC_ID starting with '11'.

    This function attempts to generate an 8-digit VC_ID beginning with '11',
    checking its uniqueness before returning it. A maximum attempt limit is
    set to prevent infinite loops in case the ID space is saturated.

    Args:
        max_attempts: The maximum number of attempts to generate a unique ID.

    Returns:
        A unique VC_ID instance if successful, None if no unique ID is found.
    """

    def create_vc_id() -> str:
        """Generate an 8-digit VC_ID starting with '11'."""
        return f"11{random.randint(100000, 999999)}"  # noqa: S311

    for _ in range(max_attempts):
        vc_id = create_vc_id()
        if is_virtual_circuit_id_available(vc_id):
            return VC_ID(vc_id)

    return None