Skip to content

Placement port

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

get_active_placement_port_subscriptions(partner_id=None)

Retrieve active Placement Port subscriptions.

Parameters:

Name Type Description Default
partner_id UUIDstr | None

The customer id of subscriptions.

None

Returns:

Type Description
list[SubscriptionModel]

A list of Subscription objects for Placement Ports.

Source code in gso/services/subscriptions/placement_port.py
def get_active_placement_port_subscriptions(partner_id: UUIDstr | None = None) -> list[SubscriptionModel]:
    """Retrieve active Placement Port subscriptions.

    Args:
        partner_id: The customer id of subscriptions.

    Returns:
        A list of Subscription objects for Placement Ports.
    """
    results = get_subscriptions(
        product_types=[ProductType.IX_PORT, ProductType.PRIVATE_PEER_PORT, ProductType.TRANSIT_PROVIDER_PORT],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id"],
        partner_id=partner_id,
    )

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

get_active_commercial_peers_linked_to_placement_port(placement_port_id)

Get all commercial peers that are connected to a given subscription.

Parameters:

Name Type Description Default
placement_port_id UUID

The ID of the placement port subscription.

required

Returns:

Type Description
list[SubscriptionModel]

A list of SubscriptionModel objects representing the commercial peers.

Source code in gso/services/subscriptions/placement_port.py
def get_active_commercial_peers_linked_to_placement_port(placement_port_id: UUID) -> list[SubscriptionModel]:
    """Get all commercial peers that are connected to a given subscription.

    Args:
        placement_port_id: The ID of the placement port subscription.

    Returns:
        A list of SubscriptionModel objects representing the commercial peers.
    """
    results = (
        query_in_use_by_subscriptions(placement_port_id)
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type == ProductType.COMMERCIAL_PEER.value,
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

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