Skip to content

Port label

Helper methods for Port Label subscriptions that make database interactions easier.

get_active_port_label_subscriptions(port_label_name, includes=None)

Retrieve active Port Label subscriptions.

Parameters:

Name Type Description Default
port_label_name ProductName

The Label Name of the Port Label.

 required
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 a given Port Label.
Source code in gso/services/subscriptions/port_label.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def get_active_port_label_subscriptions(
    port_label_name: ProductName, includes: list[str] | None = None
) -> list[SubscriptionType]:
    """Retrieve active Port Label subscriptions.

    Args:
        port_label_name: The Label Name of the Port Label.
        includes: The fields to be included in the returned Subscription objects.

    Returns:
        A list of Subscription objects for a given Port Label.
    """
    if port_label_name not in PortLabelName.values():
        msg = f"Product name {port_label_name} is not a valid Port Label."
        raise ValueError(msg)

    product_id = get_product_id_by_name(port_label_name)
    return get_subscriptions(
        product_types=[ProductType.EXPRESSROUTE_EDGE],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=includes,
        subscription_filters=[{"product_id": str(product_id)}],
    )

get_edge_ports_by_label(port_label_name)

Given a Port Label, return a list of Edge Port subscription IDs that have the given label.

Source code in gso/services/subscriptions/port_label.py 
37
38
39
40
41
42
43
def get_edge_ports_by_label(port_label_name: ProductName) -> list[UUID]:
    """Given a Port Label, return a list of Edge Port subscription IDs that have the given label."""
    port_label_subscriptions = [
        PortLabel.from_subscription(sub["subscription_id"])
        for sub in get_active_port_label_subscriptions(port_label_name=port_label_name, includes=["subscription_id"])
    ]
    return [sub.port_label.edge_port.owner_subscription_id for sub in port_label_subscriptions]