Subscriptions

A collection of methods that make interaction with coreDB more straight-forward.

This prevents someone from having to re-write database statements many times, that might turn out to be erroneous or inconsistent when not careful. These methods relate to operations on subscriptions.

get_subscriptions(product_types=None, lifecycles=None, includes=None, excludes=None, partner_id=None)

Retrieve active subscriptions for a specific product type.

Parameters:

Name	Type	Description	Default
product_types	list[ProductType] | None	The types of the product for which to retrieve subscriptions.	None
lifecycles	list[SubscriptionLifecycle] | None	The lifecycles that the products must be in.	None
includes	list[str] | None	List of fields to be included in the returned Subscription objects.	None
excludes	list[str] | None	List of fields to be excluded from the returned Subscription objects.	None
partner_id	UUIDstr | None	The customer id of subscriptions.	None

Returns:

Type	Description
list[SubscriptionType]	A list of SubscriptionType objects that match the query.

Source code in gso/services/subscriptions.py

def get_subscriptions(
    product_types: list[ProductType] | None = None,
    lifecycles: list[SubscriptionLifecycle] | None = None,
    includes: list[str] | None = None,
    excludes: list[str] | None = None,
    partner_id: UUIDstr | None = None,
) -> list[SubscriptionType]:
    """Retrieve active subscriptions for a specific product type.

    Args:
        product_types: The types of the product for which to retrieve subscriptions.
        lifecycles: The lifecycles that the products must be in.
        includes: List of fields to be included in the returned Subscription objects.
        excludes: List of fields to be excluded from the returned Subscription objects.
        partner_id: The customer id of subscriptions.

    Returns:
        A list of `SubscriptionType` objects that match the query.
    """
    if not includes:
        includes = [col.name for col in SubscriptionTable.__table__.columns]

    if excludes:
        includes = [field for field in includes if field not in excludes]

    dynamic_fields = [getattr(SubscriptionTable, field) for field in includes]

    query = db.session.query(SubscriptionTable).join(ProductTable)

    if product_types:
        query = query.filter(ProductTable.product_type.in_([str(product_type) for product_type in product_types]))

    if lifecycles:
        query = query.filter(SubscriptionTable.status.in_([str(lifecycle) for lifecycle in lifecycles]))

    if partner_id:
        query = query.filter(SubscriptionTable.customer_id == partner_id)

    results = query.with_entities(*dynamic_fields).all()

    return [dict(zip(includes, result, strict=True)) 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.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.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.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_active_switch_subscriptions(includes=None)

Retrieve active subscriptions specifically for switches.

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 switches.

Source code in gso/services/subscriptions.py

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

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

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

get_active_iptrunk_subscriptions(includes=None)

Retrieve active subscriptions specifically for IP trunks.

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 IP trunks.

Source code in gso/services/subscriptions.py

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

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

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

get_non_terminated_iptrunk_subscriptions(includes=None)

Retrieve all IP trunk subscriptions that are not terminated.

Parameters:

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

Returns:

Type	Description
list[SubscriptionType]	A list of IP trunk subscriptions.

Source code in gso/services/subscriptions.py

def get_non_terminated_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
    """Retrieve all IP trunk subscriptions that are not terminated.

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

    Returns:
        A list of IP trunk subscriptions.
    """
    return get_subscriptions(
        product_types=[ProductType.IP_TRUNK],
        lifecycles=[SubscriptionLifecycle.INITIAL, SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
        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	UUIDstr	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.py

def get_trunks_that_terminate_on_router(
    subscription_id: UUIDstr, 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(UUID(subscription_id))
        .join(ProductTable)
        .filter(
            ProductTable.product_type == ProductType.IP_TRUNK,
            SubscriptionTable.status == lifecycle_state,
        )
        .all()
    )

get_active_l3_services_linked_to_edge_port(edge_port_id)

Retrieve all active l3 core services that are on top of the given edge port.

Parameters:

Name	Type	Description	Default
edge_port_id	UUIDstr	The ID of the edge port.	required

Returns:

Type	Description
list[SubscriptionModel]	A list of active services that are on top of the edge port.

Source code in gso/services/subscriptions.py

def get_active_l3_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionModel]:
    """Retrieve all active l3 core services that are on top of the given edge port.

    Args:
        edge_port_id: The ID of the edge port.

    Returns:
        A list of active services that are on top of the edge port.
    """
    results = (
        query_in_use_by_subscriptions(UUID(edge_port_id))
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type.in_([L3_CORE_SERVICE_PRODUCT_TYPE]),
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

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

get_active_l2_circuit_services_linked_to_edge_port(edge_port_id)

Retrieve all active l2 circuit services that are on top of the given edge port.

Source code in gso/services/subscriptions.py

def get_active_l2_circuit_services_linked_to_edge_port(edge_port_id: UUIDstr) -> list[SubscriptionModel]:
    """Retrieve all active l2 circuit services that are on top of the given edge port."""
    results = (
        query_in_use_by_subscriptions(UUID(edge_port_id))
        .join(ProductTable)
        .filter(
            and_(
                ProductTable.product_type.in_([L2_CIRCUIT_PRODUCT_TYPE]),
                SubscriptionTable.status == SubscriptionLifecycle.ACTIVE,
            )
        )
        .all()
    )

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

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	UUIDstr	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.py

def get_active_vrfs_linked_to_router(router_id: UUIDstr) -> 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(UUID(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	UUIDstr	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.py

def get_active_ip_trunks_linked_to_router(router_id: UUIDstr) -> 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(UUID(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_product_id_by_name(product_name)

Retrieve the UUID of a product by its name.

Parameters:

Name	Type	Description	Default
product_name	ProductName	The name of the product.	required

Returns:

Type	Description
UUID	The UUID of the product.

Source code in gso/services/subscriptions.py

def get_product_id_by_name(product_name: ProductName) -> UUID:
    """Retrieve the UUID of a product by its name.

    Args:
        product_name: The name of the product.

    Returns:
        The UUID of the product.
    """
    return ProductTable.query.filter_by(name=product_name).first().product_id

get_active_subscriptions_by_field_and_value(field_name, field_value)

Retrieve a list of active subscriptions based on a specified field and its value.

Parameters:

Name	Type	Description	Default
field_name	str	The name of the field to filter by.	required
field_value	str	The value of the field to match.	required

Returns:

Type	Description
list[SubscriptionTable]	A list of active Subscription objects that match the criteria.

Source code in gso/services/subscriptions.py

def get_active_subscriptions_by_field_and_value(field_name: str, field_value: str) -> list[SubscriptionTable]:
    """Retrieve a list of active subscriptions based on a specified field and its value.

    Args:
        field_name: The name of the field to filter by.
        field_value: The value of the field to match.

    Returns:
        A list of active Subscription objects that match the criteria.
    """
    return (
        SubscriptionTable.query.join(ProductTable)
        .join(SubscriptionInstanceTable)
        .join(SubscriptionInstanceValueTable)
        .join(ResourceTypeTable)
        .filter(SubscriptionInstanceValueTable.value == field_value)
        .filter(ResourceTypeTable.resource_type == field_name)
        .filter(SubscriptionTable.status == SubscriptionLifecycle.ACTIVE)
        .all()
    )

get_subscription_by_process_id(process_id)

Get a subscription from a process ID.

Source code in gso/services/subscriptions.py

def get_subscription_by_process_id(process_id: str) -> SubscriptionModel | None:
    """Get a subscription from a process ID."""
    subscription_table = ProcessSubscriptionTable.query.filter(
        ProcessSubscriptionTable.process_id == process_id
    ).first()
    return SubscriptionModel.from_subscription(subscription_table.subscription_id) if subscription_table else None

get_active_insync_subscriptions()

Retrieve all subscriptions that are currently active and in sync.

Source code in gso/services/subscriptions.py

def get_active_insync_subscriptions() -> list[SubscriptionTable]:
    """Retrieve all subscriptions that are currently active and in sync."""
    return (
        SubscriptionTable.query.join(ProductTable)
        .filter(SubscriptionTable.insync.is_(True), SubscriptionTable.status == SubscriptionLifecycle.ACTIVE.value)
        .all()
    )

get_active_site_subscriptions(includes=None)

Retrieve active subscriptions specifically for sites.

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 sites.

Source code in gso/services/subscriptions.py

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

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

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

get_active_edge_port_subscriptions(partner_id=None)

Retrieve active Edge 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 Edge Ports.

Source code in gso/services/subscriptions.py

def get_active_edge_port_subscriptions(partner_id: UUIDstr | None = None) -> list[SubscriptionModel]:
    """Retrieve active Edge Port subscriptions.

    Args:
        partner_id: The customer id of subscriptions.

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

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

get_site_by_name(site_name)

Get a site by its name.

Parameters:

Name	Type	Description	Default
site_name	str	The name of the site.	required

Source code in gso/services/subscriptions.py

def get_site_by_name(site_name: str) -> Site:
    """Get a site by its name.

    Args:
        site_name: The name of the site.
    """
    subscription = get_active_subscriptions_by_field_and_value("site_name", site_name)
    if not subscription:
        msg = f"Site with name {site_name} not found."
        raise ValueError(msg)

    return Site.from_subscription(subscription[0].subscription_id)

get_all_active_sites()

Retrieve all active sites subscription together with instance values.

Returns:

Type	Description
list[dict[str, Any]]	A list of active sites with their subscription IDs and site instances.

Source code in gso/services/subscriptions.py

def get_all_active_sites() -> list[dict[str, Any]]:
    """Retrieve all active sites subscription together with instance values.

    Returns:
        A list of active sites with their subscription IDs and site instances.
    """
    return [
        {
            "subscription_id": subscription["subscription_id"],
            "site": Site.from_subscription(subscription["subscription_id"]).site,
        }
        for subscription in get_active_site_subscriptions(includes=["subscription_id"])
    ]

is_resource_type_value_unique(resource_type, value)

Check if the given value for the specified resource type is unique in the database.

This function verifies if the specified value for the given resource type is not already in the core database.

:param resource_type: The resource type to check. :type resource_type: str :param value: The value to check. :type value: str :return: True if the value is unique (not found), False if it exists. :rtype: bool

Source code in gso/services/subscriptions.py

def is_resource_type_value_unique(resource_type: str, value: str) -> bool:
    """Check if the given value for the specified resource type is unique in the database.

    This function verifies if the specified value for the given resource type is not already in the core database.

    :param resource_type: The resource type to check.
    :type resource_type: str
    :param value: The value to check.
    :type value: str
    :return: True if the value is unique (not found), False if it exists.
    :rtype: bool
    """
    exists = (
        ResourceTypeTable.query.join(SubscriptionInstanceValueTable)
        .filter(ResourceTypeTable.resource_type == resource_type, SubscriptionInstanceValueTable.value == value)
        .scalar()
    )
    return exists is None

is_virtual_circuit_id_available(virtual_circuit_id)

Check if the given virtual circuit ID is unique in the database.

This function verifies if the specified virtual circuit ID is not already present in the core database.

Parameters:

Name	Type	Description	Default
virtual_circuit_id	str	The virtual circuit ID to check.	required

Returns:

Type	Description
bool	True if the virtual circuit ID is unique (not found), False if it exists.

Source code in gso/services/subscriptions.py

def is_virtual_circuit_id_available(virtual_circuit_id: str) -> bool:
    """Check if the given virtual circuit ID is unique in the database.

    This function verifies if the specified virtual circuit ID is not already
    present in the core database.

    Args:
        virtual_circuit_id: The virtual circuit ID to check.

    Returns:
        True if the virtual circuit ID is unique (not found), False if it exists.
    """
    return is_resource_type_value_unique("virtual_circuit_id", virtual_circuit_id)

generate_unique_id(prefix)

Generate a unique ID using a shared sequence.

Parameters:

Name	Type	Description	Default
prefix	str	The prefix for the "GA" or "GS" IDs.	required

Returns:

Name	Type	Description
str	str	A unique ID in the format <prefix>-<number>.

Raises:

Type	Description
ValueError	If there is an error generating the ID.

Source code in gso/services/subscriptions.py

def generate_unique_id(prefix: Literal["GA", "GS"]) -> str:
    """Generate a unique ID using a shared sequence.

    Args:
        prefix (str): The prefix for the "GA" or "GS" IDs.

    Returns:
        str: A unique ID in the format `<prefix>-<number>`.

    Raises:
        ValueError: If there is an error generating the ID.
    """
    if prefix not in {"GA", "GS"}:
        error_message = "Invalid prefix. Only 'GA' and 'GS' are supported."
        raise ValueError(error_message)

    try:
        new_id = db.session.execute(text("SELECT nextval('ga_gs_id_seq')")).scalar_one()
    except SQLAlchemyError as exc:
        error_message = f"Error generating {prefix} ID: {exc}"
        raise ValueError(error_message) from exc
    else:
        return f"{prefix}-{new_id}"