diff --git a/gso/services/subscriptions.py b/gso/services/subscriptions.py index f4bc0b348424549ff1ed3f8184478315b0f9e303..d007cc18af22613d5544dcf84ac541b4f9396146 100644 --- a/gso/services/subscriptions.py +++ b/gso/services/subscriptions.py @@ -15,7 +15,9 @@ from orchestrator.db import ( SubscriptionInstanceValueTable, SubscriptionTable, ) +from orchestrator.services.subscriptions import query_in_use_by_subscriptions from orchestrator.types import SubscriptionLifecycle +from pydantic_forms.types import UUIDstr from gso.products import ProductType @@ -85,6 +87,38 @@ def get_active_router_subscriptions( return get_active_subscriptions(product_type="Router", includes=includes) +def get_active_iptrunk_subscriptions( + includes: list[str] | None = None, +) -> list[SubscriptionType]: + """Retrieve active subscriptions specifically for IP trunks. + + :param includes: The fields to be included in the returned Subscription objects. + :type includes: list[str] + + :return: A list of Subscription objects for IP trunks. + :rtype: list[Subscription] + """ + return get_active_subscriptions(product_type="Iptrunk", includes=includes) + + +def get_active_trunks_that_terminate_on_router(subscription_id: UUIDstr) -> list[SubscriptionTable]: + """Get all IP trunk subscriptions that are active, and terminate on the given ``subscription_id`` of a Router. + + Given a ``subscription_id`` of a Router subscription, this method gives a list of all active IP trunk subscriptions + that terminate on this Router. + + :param subscription_id: Subscription ID of a Router + :type subscription_id: UUIDstr + + :return: A list of IP trunk subscriptions + :rtype: list[SubscriptionTable] + """ + return query_in_use_by_subscriptions(UUID(subscription_id)).join(ProductTable).filter( + ProductTable.product_type == "Iptrunk", + SubscriptionTable.status == "active" + ).all() + + def get_product_id_by_name(product_name: ProductType) -> UUID: """Retrieve the :term:`UUID` of a product by its name.