Router

Helper methods for Router subscriptions that make database interactions easier.

router_site_has_another_amt_router(subscription_id)

Get the site a router is located at, and determine whether this site also houses another AMT router.

Parameters:

Name	Type	Description	Default
subscription_id	UUIDstr	The subscription ID of a Router subscription.	required

Returns:

Type	Description
bool	A boolean whether the router's site also houses an AMT router.

Source code in gso/services/subscriptions/router.py

def router_site_has_another_amt_router(subscription_id: UUIDstr) -> bool:
    """Get the site a router is located at, and determine whether this site also houses another AMT router.

    Args:
        subscription_id: The subscription ID of a Router subscription.

    Returns:
        A boolean whether the router's site also houses an AMT router.
    """
    router = Router.from_subscription(subscription_id)
    if not router:
        msg = f"Invalid router ID provided: {subscription_id}"
        raise ValueError(msg)

    non_terminated_amt_routers = [
        sub["subscription_id"]
        for sub in get_subscriptions(
            product_types=[ProductType.ROUTER],
            lifecycles=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
            exclude_subscription_ids=[subscription_id],
            instance_filters=[{"router_role": RouterRole.AMT}],
        )
    ]

    routers_on_site = [
        sub.subscription_id
        for sub in query_in_use_by_subscriptions(router.router.router_site.owner_subscription_id)
        .filter(
            and_(
                SubscriptionTable.product_id == get_product_id_by_name(ProductName.ROUTER),
                SubscriptionTable.status.in_([SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]),
            )
        )
        .all()
    ]

    return any(amt in routers_on_site for amt in non_terminated_amt_routers)

get_active_layer_3_services_on_router(subscription_id)

Get all active Layer 3 services that insist on a given router subscription_id.

Parameters:

Name	Type	Description	Default
subscription_id	UUID	Subscription ID of a Router.	required

Returns:

Type	Description
list[SubscriptionModel]	A list of Router subscriptions.

Source code in gso/services/subscriptions/router.py

def get_active_layer_3_services_on_router(subscription_id: UUID) -> list[SubscriptionModel]:
    """Get all active Layer 3 services that insist on a given router `subscription_id`.

    Args:
        subscription_id: Subscription ID of a Router.

    Returns:
        A list of Router subscriptions.
    """
    active_edge_ports = (
        query_in_use_by_subscriptions(subscription_id)
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type == ProductType.EDGE_PORT,
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

    active_l3_services = []
    for edge_port in active_edge_ports:
        active_l3_services.extend(get_active_l3_services_linked_to_edge_port(edge_port.subscription_id))

    return active_l3_services

get_active_vrfs_linked_to_router(router_id)

Retrieve all active VRFs that are linked to the router.

Parameters:

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

Returns:

Type	Description
list[SubscriptionModel]	A list of active VRFs that are linked to the router.

Source code in gso/services/subscriptions/router.py

def get_active_vrfs_linked_to_router(router_id: UUID) -> list[SubscriptionModel]:
    """Retrieve all active VRFs that are linked to the router.

    Args:
        router_id: The ID of the router.

    Returns:
        A list of active VRFs that are linked to the router.
    """
    results = (
        query_in_use_by_subscriptions(router_id)
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type == ProductType.VRF.value,
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

    return [SubscriptionModel.from_subscription(result.subscription_id) for result in results]

get_active_ip_trunks_linked_to_router(router_id)

Retrieve all active IP trunks that are linked to the router.

Parameters:

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

Returns:

Type	Description
list[SubscriptionModel]	A list of active IP trunks that are linked to the router.

Source code in gso/services/subscriptions/router.py

def get_active_ip_trunks_linked_to_router(router_id: UUID) -> list[SubscriptionModel]:
    """Retrieve all active IP trunks that are linked to the router.

    Args:
        router_id: The ID of the router.

    Returns:
        A list of active IP trunks that are linked to the router.
    """
    results = (
        query_in_use_by_subscriptions(router_id)
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type == ProductType.IP_TRUNK.value,
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

    return [SubscriptionModel.from_subscription(result.subscription_id) for result in results]

get_router_subscriptions(includes=None, lifecycles=None)

Retrieve subscriptions specifically for routers.

Parameters:

Name	Type	Description	Default
includes	list[str] | None	The fields to be included in the returned Subscription objects.	None
lifecycles	list[SubscriptionLifecycle] | None	The subscription lifecycle states that should be included in the results.	None

Returns:

Type	Description
list[SubscriptionType]	A list of Subscription objects for routers.

Source code in gso/services/subscriptions/router.py

def get_router_subscriptions(
    includes: list[str] | None = None, lifecycles: list[SubscriptionLifecycle] | None = None
) -> list[SubscriptionType]:
    """Retrieve subscriptions specifically for routers.

    Args:
        includes: The fields to be included in the returned Subscription objects.
        lifecycles: The subscription lifecycle states that should be included in the results.

    Returns:
        A list of Subscription objects for routers.
    """
    return get_subscriptions(product_types=[ProductType.ROUTER], lifecycles=lifecycles, includes=includes)

get_active_router_subscriptions(includes=None)

Retrieve active subscriptions specifically for routers.

Parameters:

Name	Type	Description	Default
includes	list[str] | None	The fields to be included in the returned Subscription objects.	None

Returns:

Type	Description
list[SubscriptionType]	A list of Subscription objects for routers.

Source code in gso/services/subscriptions/router.py

def get_active_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
    """Retrieve active subscriptions specifically for routers.

    Args:
        includes: The fields to be included in the returned Subscription objects.

    Returns:
        A list of Subscription objects for routers.
    """
    return get_subscriptions(
        product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
    )

get_provisioning_router_subscriptions(includes=None)

Retrieve provisioning subscriptions specifically for routers.

Parameters:

Name	Type	Description	Default
includes	list[str] | None	The fields to be included in the returned Subscription objects.	None

Returns:

Type	Description
list[SubscriptionType]	A list of router Subscription objects.

Source code in gso/services/subscriptions/router.py

def get_provisioning_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
    """Retrieve provisioning subscriptions specifically for routers.

    Args:
        includes: The fields to be included in the returned Subscription objects.

    Returns:
        A list of router Subscription objects.
    """
    return get_subscriptions(
        product_types=[ProductType.ROUTER], lifecycles=[SubscriptionLifecycle.PROVISIONING], includes=includes
    )

get_trunks_that_terminate_on_router(subscription_id, lifecycle_state)

Get all IP trunk subscriptions that terminate on the given subscription_id of a Router.

Given a subscription_id of a Router subscription, this method gives a list of all IP trunk subscriptions that terminate on this Router. The given lifecycle state dictates the state of trunk subscriptions that are counted as terminating on this router.

Parameters:

Name	Type	Description	Default
subscription_id	UUID	Subscription ID of a Router	required
lifecycle_state	SubscriptionLifecycle	Required lifecycle state of the IP trunk	required

Returns:

Type	Description
list[SubscriptionTable]	A list of IP trunk subscriptions

Source code in gso/services/subscriptions/router.py

def get_trunks_that_terminate_on_router(
    subscription_id: UUID, lifecycle_state: SubscriptionLifecycle
) -> list[SubscriptionTable]:
    """Get all IP trunk subscriptions that terminate on the given `subscription_id` of a Router.

    Given a `subscription_id` of a Router subscription, this method gives a list of all IP trunk subscriptions that
    terminate on this Router. The given lifecycle state dictates the state of trunk subscriptions that are counted as
    terminating on this router.

    Args:
        subscription_id: Subscription ID of a Router
        lifecycle_state: Required lifecycle state of the IP trunk

    Returns:
        A list of IP trunk subscriptions
    """
    return (
        query_in_use_by_subscriptions(subscription_id)
        .join(ProductTable)
        .filter(and_(ProductTable.product_type == ProductType.IP_TRUNK, SubscriptionTable.status == lifecycle_state))
        .all()
    )

get_active_pe_routers(includes=None, exclude_subscription_ids=None, vendor=None)

Retrieve active PE routers.

Source code in gso/services/subscriptions/router.py

def get_active_pe_routers(
    includes: list[str] | None = None,
    exclude_subscription_ids: list[UUIDstr] | None = None,
    vendor: Vendor | None = None,
) -> list[SubscriptionType]:
    """Retrieve active PE routers."""
    instance_filters = [{"router_role": RouterRole.PE.value}]
    if vendor:
        instance_filters.append({"vendor": vendor.value})

    return get_subscriptions(
        product_types=[ProductType.ROUTER],
        includes=includes,
        instance_filters=instance_filters,
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        exclude_subscription_ids=exclude_subscription_ids,
    )

get_active_nokia_router_subscriptions(includes=None)

Retrieve active subscriptions specifically for routers.

Parameters:

Name	Type	Description	Default
includes	list[str] | None	The fields to be included in the returned Subscription objects.	None

Returns:

Type	Description
list[SubscriptionType]	A list of Subscription objects for routers.

Source code in gso/services/subscriptions/router.py

def get_active_nokia_router_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
    """Retrieve active subscriptions specifically for routers.

    Args:
        includes: The fields to be included in the returned Subscription objects.

    Returns:
        A list of Subscription objects for routers.
    """
    return get_subscriptions(
        product_types=[ProductType.ROUTER],
        lifecycles=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
        includes=includes,
        instance_filters=[{"vendor": Vendor.NOKIA}],
    )