Product types and block for 10GGBS

gso/migrations/versions/2025-02-18_3f54c454f4bf_add_10ggbs_product.py

+"""Add 10GGBS product.
+
+Revision ID: 3f54c454f4bf
+Revises: efebcde91f2f
+Create Date: 2025-02-18 14:02:54.974421
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '3f54c454f4bf'
+down_revision = 'efebcde91f2f'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('10GGBS', '10GGBS product', 'TenGGBS', '10GGBS', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Imported 10GGBS', 'Imported 10GGBS product', 'ImportedTenGGBS', 'IMP_10GGBS', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('TenGGBSBlock', '10GGBS product block', '10GGBS_BLK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('PathBlock', '10GGBS product path block', '10GGBS_PATH_BLK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('path_role', 'The role of the path (primary or secondary) for 10GGBS service') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_product_blocks (product_id, product_block_id) VALUES ((SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock'))), ((SELECT products.product_id FROM products WHERE products.name IN ('10GGBS')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_relations (in_use_by_id, depends_on_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('IptrunkSideBlock')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_block_resource_types (product_block_id, resource_type_id) VALUES ((SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('path_role')))
+    """))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    conn.execute(sa.text("""
+DELETE FROM product_block_resource_types WHERE product_block_resource_types.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('path_role'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values USING product_block_resource_types WHERE subscription_instance_values.subscription_instance_id IN (SELECT subscription_instances.subscription_instance_id FROM subscription_instances WHERE subscription_instances.subscription_instance_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('path_role'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instance_values WHERE subscription_instance_values.resource_type_id IN (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('path_role'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('path_role')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_product_blocks WHERE product_product_blocks.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS')) AND product_product_blocks.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('TenGGBSBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_block_relations WHERE product_block_relations.in_use_by_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('IptrunkSideBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.product_block_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('PathBlock', 'TenGGBSBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_blocks WHERE product_blocks.name IN ('PathBlock', 'TenGGBSBlock')
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes WHERE processes.pid IN (SELECT processes_subscriptions.pid FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS'))))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM processes_subscriptions WHERE processes_subscriptions.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscription_instances WHERE subscription_instances.subscription_id IN (SELECT subscriptions.subscription_id FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM products WHERE products.name IN ('Imported 10GGBS', '10GGBS')
+    """))

gso/products/__init__.py

@@ -20,8 +20,8 @@ from gso.products.product_types.router import ImportedRouter, Router
 from gso.products.product_types.site import ImportedSite, Site
 from gso.products.product_types.super_pop_switch import ImportedSuperPopSwitch, SuperPopSwitch
 from gso.products.product_types.switch import ImportedSwitch, Switch
+from gso.products.product_types.tenggbs import ImportedTenGGBS, TenGGBS
 from gso.products.product_types.vrf import VRF
-from gso.products.product_types.tenggbs import TenGGBS, ImportedTenGGBS
 
 
 class ProductName(strEnum):
@@ -79,9 +79,9 @@ class ProductName(strEnum):
     IMPORTED_EXPRESSROUTE = Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE
     VRF = "VRF"
     """VRFs."""
-    TenGGBS = "TenGGBS"
+    TENGGBS = "10GGBS"
     """10GGBS."""
-    IMPORTED_TenGGBS = "Imported TenGGBS"
+    IMPORTED_TENGGBS = "Imported 10GGBS"
 
 
 L3_CORE_SERVICE_PRODUCT_TYPE = L3CoreService.__name__
@@ -165,8 +165,8 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.EXPRESSROUTE.value: Layer2Circuit,
         ProductName.IMPORTED_EXPRESSROUTE.value: ImportedLayer2Circuit,
         ProductName.VRF.value: VRF,
-        ProductName.TenGGBS.value: TenGGBS,
-        ProductName.IMPORTED_TenGGBS.value: ImportedTenGGBS,
+        ProductName.TENGGBS.value: TenGGBS,
+        ProductName.IMPORTED_TENGGBS.value: ImportedTenGGBS,
     },
 )
 

gso/products/product_blocks/tenggbs.py

 """10GGBS product block models."""
+
 from typing import Annotated, TypeVar
 
 from annotated_types import Len
 from orchestrator.domain.base import ProductBlockModel
-from orchestrator.types import SubscriptionLifecycle
-from orchestrator.types import strEnum
+from orchestrator.types import SubscriptionLifecycle, strEnum
 from pydantic import AfterValidator
 from pydantic_forms.validators import validate_unique_list
 from typing_extensions import Doc
@@ -25,6 +25,7 @@ T = TypeVar("T")
 
 class PathRole(strEnum):
     """Defines path role types."""
+
     PRIMARY = "Primary"
     """Primary path."""
     SECONDARY = "Secondary"
@@ -34,37 +35,40 @@ class PathRole(strEnum):
 Paths = Annotated[
     list[T],
     AfterValidator(validate_unique_list),
-    Len(min_length=0),
     Doc("A list of TrunkSideBlocks that make up a path."),
 ]
 
 
-class PathBlockInactive(
-    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="PathBlock"
-):
+class PathBlockInactive(ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="PathBlock"):
     """A path representing an ordered list of adjacent trunk sides."""
 
-    role: PathRole | None = None
-    path: Paths[IptrunkSideBlockInactive]
+    path_role: PathRole | None = None
+    tenggbs_path: Paths[IptrunkSideBlockInactive]
 
 
 class PathBlockProvisioning(PathBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """A path that is being provisioned."""
 
-    role: PathRole | None = None
-    path: Paths[IptrunkSideBlockProvisioning]
+    path_role: PathRole | None = None
+    tenggbs_path: Paths[IptrunkSideBlockProvisioning]  # type: ignore[assignment]
 
 
 class PathBlock(PathBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
-    """An active path."""
+    """An active path for a 10GGBS service.
 
-    role: PathRole
-    path: Paths[IptrunkSideBlock]
+    Attributes:
+        path_role: The role of the path (primary or secondary).
+        tenggbs_path: The list of trunk sides that make up the path.
+    """
+
+    path_role: PathRole
+    tenggbs_path: Paths[IptrunkSideBlock]  # type: ignore[assignment]
 
 
 PathsList = Annotated[
     list[T],
     AfterValidator(validate_unique_list),
+    Len(min_length=0),
     Doc("A list of paths (1 primary, multiple secondary)."),
 ]
 
@@ -75,18 +79,23 @@ class TenGGBSBlockInactive(
     """Inactive 10GGBS service model."""
 
     l2_circuit: Layer2CircuitBlockInactive
-    paths: PathsList[PathBlockInactive] | None = None
+    tenggbs_paths: PathsList[PathBlockInactive]
 
 
 class TenGGBSBlockProvisioning(TenGGBSBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """Provisioning state of 10GGBS."""
 
     l2_circuit: Layer2CircuitBlockProvisioning
-    paths: PathsList[PathBlockProvisioning] | None = None
+    tenggbs_paths: PathsList[PathBlockProvisioning]  # type: ignore[assignment]
 
 
 class TenGGBSBlock(TenGGBSBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
-    """Active 10GGBS service model."""
+    """Active 10GGBS service model.
+
+    Attributes:
+        l2_circuit: The Layer 2 circuit block.
+        tenggbs_paths: The list of paths for this 10GGBS service.
+    """
 
     l2_circuit: Layer2CircuitBlock
-    paths: PathsList[PathBlock] | None = None
+    tenggbs_paths: PathsList[PathBlock]  # type: ignore[assignment]

gso/products/product_types/tenggbs.py

 """The product type for 10GGBS services."""
 
-from orchestrator.domain.base import SubscriptionModel
+from orchestrator.domain import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle
 
 from gso.products.product_blocks.tenggbs import (