gso/cli/imports.py

@@ -22,7 +22,6 @@ from gso.products.product_blocks.bgp_session import IPFamily
 from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.service_binding_port import VLAN_ID
 from gso.products.product_blocks.switch import SwitchModel
 from gso.products.product_types.nren_l3_core_service import NRENL3CoreServiceType
 from gso.services.partners import (
@@ -50,6 +49,7 @@ from gso.utils.types.ip_address import (
     IPV6Netmask,
     PortNumber,
 )
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 app: typer.Typer = typer.Typer()
 

gso/migrations/versions/2024-10-29_5132c463214d_add_layer_2_cricuits_services_domain_.py

+"""Add Layer 2 Cricuits services domain models..
+
+Revision ID: 5132c463214d
+Revises: 0e7e7d749617
+Create Date: 2024-10-29 13:34:39.234303
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '5132c463214d'
+down_revision = '0e7e7d749617'
+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 ('GÉANT Plus', 'GÉANT Plus', 'Layer2Circuit', 'G_PLUS', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Imported GÉANT Plus', 'Imported GEANT Plus', 'ImportedLayer2Circuit', 'IMP_G_PLUS', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Azure ExpressRoute', 'Azure ExpressRoute product', 'Layer2Circuit', 'ER', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO products (name, description, product_type, tag, status) VALUES ('Imported Azure ExpressRoute', 'Imported Azure ExpressRoute', 'ImportedLayer2Circuit', 'IMP_ER', 'active') RETURNING products.product_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO fixed_inputs (name, value, product_id) VALUES ('layer_2_circuit_service_type', 'Azure ExpressRoute', (SELECT products.product_id FROM products WHERE products.name IN ('Azure ExpressRoute'))), ('layer_2_circuit_service_type', 'Imported Azure ExpressRoute', (SELECT products.product_id FROM products WHERE products.name IN ('Imported Azure ExpressRoute'))), ('layer_2_circuit_service_type', 'Imported GÉANT Plus', (SELECT products.product_id FROM products WHERE products.name IN ('Imported GÉANT Plus'))), ('layer_2_circuit_service_type', 'GÉANT Plus', (SELECT products.product_id FROM products WHERE products.name IN ('GÉANT Plus')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('Layer2CircuitBlock', 'Layer 2 Circuit product block', 'L2_C_BLOCK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO product_blocks (name, description, tag, status) VALUES ('Layer2CircuitSideBlock', 'Layer 2 Circuit side product block', 'L2_C_SIDE_BLOCK', 'active') RETURNING product_blocks.product_block_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('policer_enabled', 'Whether this Layer 2 Circuit is policed.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('bandwidth', 'If policed, the bandwidth of the policer is stored.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('vlan_range_upper_bound', 'VLAN upper bound range') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('vlan_range_lower_bound', 'VLAN lower bound range') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('virtual_circuit_id', 'Virtual Circuit ID of this Layer 2 Circuit.') RETURNING resource_types.resource_type_id
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('layer_2_circuit_type', 'The type of circuit, can be tagged or untagged.') 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 ('Azure ExpressRoute')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitBlock'))), ((SELECT products.product_id FROM products WHERE products.name IN ('Imported Azure ExpressRoute')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitBlock'))), ((SELECT products.product_id FROM products WHERE products.name IN ('Imported GÉANT Plus')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitBlock'))), ((SELECT products.product_id FROM products WHERE products.name IN ('GÉANT Plus')), (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 ('Layer2CircuitBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitSideBlock')))
+    """))
+    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 ('Layer2CircuitSideBlock')), (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('virtual_circuit_id')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('layer_2_circuit_type')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_lower_bound')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_upper_bound')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_enabled')))
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('bandwidth')))
+    """))
+    conn.execute(sa.text("""
+INSERT INTO resource_types (resource_type, description) VALUES ('policer_burst_rate', 'Max burst rate of a policer') RETURNING resource_types.resource_type_id
+    """))
+    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 ('Layer2CircuitBlock')), (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_burst_rate')))
+    """))
+
+
+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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_burst_rate'))
+            """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_burst_rate'))
+            """))
+    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 ('policer_burst_rate'))
+            """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('policer_burst_rate')
+            """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('virtual_circuit_id'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('virtual_circuit_id'))
+    """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('layer_2_circuit_type'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('layer_2_circuit_type'))
+    """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_lower_bound'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_lower_bound'))
+    """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_upper_bound'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('vlan_range_upper_bound'))
+    """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_enabled'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('policer_enabled'))
+    """))
+    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 ('Layer2CircuitBlock')) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('bandwidth'))
+    """))
+    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 ('Layer2CircuitBlock'))) AND product_block_resource_types.resource_type_id = (SELECT resource_types.resource_type_id FROM resource_types WHERE resource_types.resource_type IN ('bandwidth'))
+    """))
+    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 ('policer_enabled', 'bandwidth', 'vlan_range_upper_bound', 'vlan_range_lower_bound', 'virtual_circuit_id', 'layer_2_circuit_type'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM resource_types WHERE resource_types.resource_type IN ('policer_enabled', 'bandwidth', 'vlan_range_upper_bound', 'vlan_range_lower_bound', 'virtual_circuit_id', 'layer_2_circuit_type')
+    """))
+    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 ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus')) AND product_product_blocks.product_block_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 ('Layer2CircuitBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitSideBlock'))
+    """))
+    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 ('Layer2CircuitSideBlock')) AND product_block_relations.depends_on_id IN (SELECT product_blocks.product_block_id FROM product_blocks WHERE product_blocks.name IN ('ServiceBindingPort'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM fixed_inputs WHERE fixed_inputs.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus')) AND fixed_inputs.name = 'layer_2_circuit_service_type'
+    """))
+    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 ('Layer2CircuitSideBlock', 'Layer2CircuitBlock'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM product_blocks WHERE product_blocks.name IN ('Layer2CircuitSideBlock', 'Layer2CircuitBlock')
+    """))
+    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 ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus'))))
+    """))
+    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 ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus')))
+    """))
+    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 ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus')))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM subscriptions WHERE subscriptions.product_id IN (SELECT products.product_id FROM products WHERE products.name IN ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus'))
+    """))
+    conn.execute(sa.text("""
+DELETE FROM products WHERE products.name IN ('Azure ExpressRoute', 'Imported Azure ExpressRoute', 'Imported GÉANT Plus', 'GÉANT Plus')
+    """))

gso/migrations/versions/2024-10-29_72a4f7aa499d_add_l2circuit_workflows.py

+"""Add L2Circuit workflows..
+
+Revision ID: 72a4f7aa499d
+Revises: 5132c463214d
+Create Date: 2024-10-29 14:04:29.807253
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '72a4f7aa499d'
+down_revision = '5132c463214d'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "create_layer_2_circuit",
+        "target": "CREATE",
+        "description": "Create Layer 2 Circuit Service",
+        "product_type": "Layer2Circuit"
+    },
+    {
+        "name": "modify_layer_2_circuit",
+        "target": "MODIFY",
+        "description": "Modify Layer 2 Circuit Service",
+        "product_type": "Layer2Circuit"
+    },
+    {
+        "name": "terminate_layer_2_circuit",
+        "target": "TERMINATE",
+        "description": "Terminate Layer 2 Circuit Service",
+        "product_type": "Layer2Circuit"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/products/__init__.py

@@ -11,6 +11,7 @@ from pydantic_forms.types import strEnum
 from gso.products.product_types.edge_port import EdgePort, ImportedEdgePort
 from gso.products.product_types.iptrunk import ImportedIptrunk, Iptrunk
 from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnect, LanSwitchInterconnect
+from gso.products.product_types.layer_2_circuit import ImportedLayer2Circuit, Layer2Circuit, Layer2CircuitServiceType
 from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreService
 from gso.products.product_types.office_router import ImportedOfficeRouter, OfficeRouter
 from gso.products.product_types.opengear import ImportedOpengear, Opengear
@@ -47,6 +48,10 @@ class ProductName(strEnum):
     IMPORTED_GEANT_IP = "Imported GÉANT IP"
     IAS = "IAS"
     IMPORTED_IAS = "Imported IAS"
+    GEANT_PLUS = Layer2CircuitServiceType.GEANT_PLUS
+    IMPORTED_GEANT_PLUS = Layer2CircuitServiceType.IMPORTED_GEANT_PLUS
+    EXPRESSROUTE = Layer2CircuitServiceType.EXPRESSROUTE
+    IMPORTED_EXPRESSROUTE = Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE
 
 
 class ProductType(strEnum):
@@ -75,6 +80,10 @@ class ProductType(strEnum):
     IMPORTED_GEANT_IP = ImportedNRENL3CoreService.__name__
     IAS = NRENL3CoreService.__name__
     IMPORTED_IAS = ImportedNRENL3CoreService.__name__
+    GEANT_PLUS = Layer2Circuit.__name__
+    IMPORTED_GEANT_PLUS = ImportedLayer2Circuit.__name__
+    EXPRESSROUTE = Layer2Circuit.__name__
+    IMPORTED_EXPRESSROUTE = ImportedLayer2Circuit.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -102,5 +111,11 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.IMPORTED_GEANT_IP.value: ImportedNRENL3CoreService,
         ProductName.IAS.value: NRENL3CoreService,
         ProductName.IMPORTED_IAS.value: ImportedNRENL3CoreService,
+        ProductName.GEANT_PLUS.value: Layer2Circuit,
+        ProductName.IMPORTED_GEANT_PLUS.value: ImportedLayer2Circuit,
+        ProductName.EXPRESSROUTE.value: Layer2Circuit,
+        ProductName.IMPORTED_EXPRESSROUTE.value: ImportedLayer2Circuit,
     },
 )
+
+__all__ = ["ProductName", "ProductType"]

gso/products/product_blocks/layer_2_circuit.py

+"""Layer 2 Circuit product block."""
+
+from collections.abc import Sequence
+from typing import Annotated, TypeVar
+
+from annotated_types import Len
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic import AfterValidator
+from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
+from typing_extensions import Doc
+
+from gso.products.product_blocks.service_binding_port import (
+    ServiceBindingPort,
+    ServiceBindingPortInactive,
+    ServiceBindingPortProvisioning,
+)
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
+
+
+class Layer2CircuitSideBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="Layer2CircuitSideBlock",
+):
+    """One inactive side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPortInactive
+
+
+class Layer2CircuitSideBlockProvisioning(
+    Layer2CircuitSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """One provisioning side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPortProvisioning
+
+
+class Layer2CircuitSideBlock(Layer2CircuitSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """One side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPort
+
+
+Layer2CircuitSideBlockType = TypeVar(
+    "Layer2CircuitSideBlockType",
+    "Layer2CircuitSideBlockInactive",
+    "Layer2CircuitSideBlockProvisioning",
+    "Layer2CircuitSideBlock",
+)
+
+Layer2CircuitSides = Annotated[
+    Sequence[Layer2CircuitSideBlockType],
+    AfterValidator(validate_unique_list),
+    Len(min_length=2, max_length=2),
+    Doc("A list of two Layer 2 Circuit sides."),
+]
+
+
+class Layer2CircuitType(strEnum):
+    """The two types of Layer 2 Circuit."""
+
+    TAGGED = "TAGGED"
+    UNTAGGED = "UNTAGGED"
+
+
+class Layer2CircuitBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="Layer2CircuitBlock"
+):
+    """An inactive Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockInactive]
+    virtual_circuit_id: VC_ID | None = None
+    layer_2_circuit_type: Layer2CircuitType | None = None
+    vlan_range_lower_bound: VLAN_ID | None = None
+    vlan_range_upper_bound: VLAN_ID | None = None
+    policer_enabled: bool | None = None
+    policer_burst_rate: BandwidthString | None = None
+    bandwidth: BandwidthString | None = None
+
+
+class Layer2CircuitBlockProvisioning(Layer2CircuitBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A provisioning Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockProvisioning]
+    virtual_circuit_id: VC_ID
+    layer_2_circuit_type: Layer2CircuitType
+    vlan_range_lower_bound: VLAN_ID | None
+    vlan_range_upper_bound: VLAN_ID | None
+    policer_enabled: bool
+    policer_burst_rate: BandwidthString | None
+    bandwidth: BandwidthString | None
+
+
+class Layer2CircuitBlock(Layer2CircuitBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active Layer 2 Circuit."""
+
+    #: The two sides that the Layer 2 Circuit is connected to.
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlock]
+    #: Virtual Circuit ID of this Layer 2 Circuit.
+    virtual_circuit_id: VC_ID
+    #: The type of circuit, can be tagged or untagged.
+    layer_2_circuit_type: Layer2CircuitType
+    #: If tagged, the lower and upper bounds will set the :term:`VLAN` range.
+    vlan_range_lower_bound: VLAN_ID | None
+    #: Lower and Upper bounds are including.
+    vlan_range_upper_bound: VLAN_ID | None
+    #: Whether this Layer 2 Circuit is policed.
+    policer_enabled: bool
+    #: If policed, the burst rate of the policer
+    policer_burst_rate: BandwidthString | None
+    #: If policed, the bandwidth of the policer is stored.
+    bandwidth: BandwidthString | None

gso/products/product_blocks/service_binding_port.py

@@ -3,8 +3,6 @@
 A service binding port is used to logically attach an edge port to a customer service using a :term:`VLAN`.
 """
 
-from typing import Annotated
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import Field
@@ -13,8 +11,7 @@ from gso.products.product_blocks.bgp_session import BGPSession, BGPSessionInacti
 from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
 from gso.utils.shared_enums import SBPType
 from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
-
-VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
 class ServiceBindingPortInactive(

gso/products/product_types/layer_2_circuit.py

+"""Product type for a Layer 2 circuit."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
+
+from gso.products.product_blocks.layer_2_circuit import (
+    Layer2CircuitBlock,
+    Layer2CircuitBlockInactive,
+    Layer2CircuitBlockProvisioning,
+)
+
+
+class Layer2CircuitServiceType(strEnum):
+    """Available types of Layer 2 Circuit services."""
+
+    GEANT_PLUS = "GÉANT Plus"
+    IMPORTED_GEANT_PLUS = "Imported GÉANT Plus"
+    EXPRESSROUTE = "Azure ExpressRoute"
+    IMPORTED_EXPRESSROUTE = "Imported Azure ExpressRoute"
+
+
+class Layer2CircuitInactive(SubscriptionModel, is_base=True):
+    """An inactive Layer 2 Circuit."""
+
+    layer_2_circuit_service_type: Layer2CircuitServiceType
+    layer_2_circuit: Layer2CircuitBlockInactive
+
+
+class Layer2CircuitProvisioning(Layer2CircuitInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Layer 2 Circuit that is provisioning."""
+
+    layer_2_circuit_service_type: Layer2CircuitServiceType
+    layer_2_circuit: Layer2CircuitBlockProvisioning
+
+
+class Layer2Circuit(Layer2CircuitProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active Layer 2 Circuit."""
+
+    layer_2_circuit_service_type: Layer2CircuitServiceType
+    layer_2_circuit: Layer2CircuitBlock
+
+
+class ImportedLayer2CircuitInactive(SubscriptionModel, is_base=True):
+    """An imported, inactive Layer 2 Circuit."""
+
+    layer_2_circuit_service_type: Layer2CircuitServiceType
+    layer_2_circuit: Layer2CircuitBlockInactive
+
+
+class ImportedLayer2Circuit(
+    ImportedLayer2CircuitInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
+):
+    """An imported Layer 2 Circuit."""
+
+    layer_2_circuit_service_type: Layer2CircuitServiceType
+    layer_2_circuit: Layer2CircuitBlock

gso/services/lso_client.py

@@ -10,11 +10,11 @@ from typing import Any, Literal, TypedDict
 import requests
 from orchestrator import step
 from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
 from orchestrator.types import State
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import Step, StepList, begin, callback_step, conditional, inputstep
 from pydantic import ConfigDict
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label, LongText, ReadOnlyField
 

gso/services/subscriptions.py

@@ -287,3 +287,25 @@ def get_all_active_sites() -> list[dict[str, Any]]:
         }
         for subscription in get_active_site_subscriptions(includes=["subscription_id"])
     ]
+
+
+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.
+
+    :param virtual_circuit_id: The virtual circuit ID to check.
+    :type virtual_circuit_id: str
+    :return: True if the virtual circuit ID is unique (not found), False if it exists.
+    :rtype: bool
+    """
+    exists = (
+        ResourceTypeTable.query.join(SubscriptionInstanceValueTable)
+        .filter(
+            ResourceTypeTable.resource_type == "virtual_circuit_id",
+            SubscriptionInstanceValueTable.value == virtual_circuit_id,
+        )
+        .scalar()
+    )
+    return exists is None

gso/utils/helpers.py

 """Helper methods that are used across :term:`GSO`."""
 
+import random
 import re
 from typing import TYPE_CHECKING
 from uuid import UUID
@@ -13,9 +14,11 @@ from gso.products.product_types.router import Router
 from gso.services import subscriptions
 from gso.services.netbox_client import NetboxClient
 from gso.services.partners import get_all_partners
+from gso.services.subscriptions import is_virtual_circuit_id_available
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 from gso.utils.types.ip_address import IPv4AddressType
+from gso.utils.types.virtual_identifiers import VC_ID
 
 if TYPE_CHECKING:
     from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
@@ -248,3 +251,28 @@ def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int
     if number_of_members > 1 and not enable_lacp:
         err_msg = "Number of members must be 1 if LACP is disabled."
         raise ValueError(err_msg)
+
+
+def generate_unique_vc_id(max_attempts: int = 100) -> VC_ID | None:
+    """Generate a unique 8-digit VC_ID starting with '11'.
+
+    This function attempts to generate an 8-digit VC_ID beginning with '11',
+    checking its uniqueness before returning it. A maximum attempt limit is
+    set to prevent infinite loops in case the ID space is saturated.
+
+    :param max_attempts: The maximum number of attempts to generate a unique ID.
+    :type max_attempts: int
+    :return: A unique VC_ID instance if successful, None if no unique ID is found.
+    :rtype: Optional[VC_ID]
+    """
+
+    def create_vc_id() -> str:
+        """Generate an 8-digit VC_ID starting with "11"."""
+        return f"11{random.randint(100000, 999999)}"  # noqa: S311
+
+    for _ in range(max_attempts):
+        vc_id = create_vc_id()
+        if is_virtual_circuit_id_available(vc_id):
+            return VC_ID(vc_id)
+
+    return None

gso/utils/types/interfaces.py

@@ -100,3 +100,26 @@ class PhysicalPortCapacity(strEnum):
     TEN_GIGABIT_PER_SECOND = "10G"
     HUNDRED_GIGABIT_PER_SECOND = "100G"
     FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
+
+
+def bandwidth_string_is_valid(bandwidth_string: str) -> str:
+    """Expect a bandwidth definition to follow the pattern of an int followed by a single letter.
+
+    If this string does not consist of a number followed by a single
+    """
+    msg = f"Expected a network capacity, e.g. 40G or 200M. Got: {bandwidth_string}"
+    if len(bandwidth_string) < 2:  # noqa: PLR2004 not a magic value
+        raise ValueError(msg)
+
+    if bandwidth_string[-1:] not in "K" "M" "G" "T":
+        raise ValueError(msg)
+
+    try:
+        int(bandwidth_string[:-1])  # Try parsing the bandwidth number
+    except ValueError as e:
+        raise ValueError(msg) from e
+
+    return bandwidth_string
+
+
+BandwidthString = Annotated[str, AfterValidator(bandwidth_string_is_valid)]

gso/utils/types/virtual_identifiers.py

+"""Annotated types for virtual identifiers such as :term:`VLAN` ID or Virtual Circuit ID."""
+
+from typing import Annotated
+
+from pydantic import Field
+from typing_extensions import Doc
+
+VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
+VC_ID = Annotated[
+    int,
+    Field(gt=0, le=2147483648),
+    Doc(
+        "A Virtual Circuit ID, the upper limit comes from the highest number that a service ID could be in Nokia srOS."
+    ),
+]

gso/utils/workflow_steps.py

@@ -5,12 +5,12 @@ from typing import Any
 
 from orchestrator import inputstep, step
 from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
 from orchestrator.types import State, UUIDstr
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, conditional
 from pydantic import ConfigDict
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label
 

gso/workflows/__init__.py

@@ -124,3 +124,8 @@ LazyWorkflowInstance(
 )
 LazyWorkflowInstance("gso.workflows.nren_l3_core_service.import_nren_l3_core_service", "import_nren_l3_core_service")
 LazyWorkflowInstance("gso.workflows.nren_l3_core_service.migrate_nren_l3_core_service", "migrate_nren_l3_core_service")
+
+# Layer 2 Circuit workflows
+LazyWorkflowInstance("gso.workflows.l2_circuit.create_layer_2_circuit", "create_layer_2_circuit")
+LazyWorkflowInstance("gso.workflows.l2_circuit.modify_layer_2_circuit", "modify_layer_2_circuit")
+LazyWorkflowInstance("gso.workflows.l2_circuit.terminate_layer_2_circuit", "terminate_layer_2_circuit")

gso/workflows/edge_port/create_imported_edge_port.py

@@ -15,7 +15,7 @@ from pydantic_forms.validators import validate_unique_list
 
 from gso.products import ProductName
 from gso.products.product_blocks.edge_port import EdgePortAEMemberBlockInactive, EdgePortType, EncapsulationType
-from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
+from gso.products.product_types.edge_port import ImportedEdgePortInactive
 from gso.products.product_types.router import Router
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
@@ -30,17 +30,14 @@ def create_subscription(partner: str) -> State:
     product_id = get_product_id_by_name(ProductName.IMPORTED_EDGE_PORT)
     subscription = ImportedEdgePortInactive.from_product_id(product_id, partner_id)
 
-    return {
-        "subscription": subscription,
-        "subscription_id": subscription.subscription_id,
-    }
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
 
 
 def initial_input_form_generator() -> FormGenerator:
     """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
 
     class ImportEdgePort(FormPage):
-        model_config = ConfigDict(title="Import Router")
+        model_config = ConfigDict(title="Import Edge Port")
 
         node: active_pe_router_selector()  # type: ignore[valid-type]
         partner: str
@@ -63,7 +60,7 @@ def initial_input_form_generator() -> FormGenerator:
 
 @step("Initialize subscription")
 def initialize_subscription(
-    subscription: EdgePortInactive,
+    subscription: ImportedEdgePortInactive,
     node: UUIDstr,
     service_type: EdgePortType,
     speed: PhysicalPortCapacity,

gso/workflows/l2_circuit/__init__.py

+"""Workflows for Layer 2 Circuits."""

gso/workflows/l2_circuit/create_imported_layer_2_circuit.py

+"""A creation workflow that adds an existing Layer 2 Circuit to the database."""
+
+from typing import Self
+
+from orchestrator import step
+from orchestrator.forms import FormPage
+from orchestrator.types import FormGenerator, State
+from pydantic import BaseModel, ConfigDict, model_validator
+from pydantic_forms.types import UUIDstr
+
+from gso.products import ProductName
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.products.product_types.layer_2_circuit import (
+    ImportedLayer2Circuit,
+    ImportedLayer2CircuitInactive,
+    Layer2CircuitServiceType,
+)
+from gso.services.partners import get_partner_by_name
+from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
+
+
+def initial_input_form_generator() -> FormGenerator:
+    """Generate a form that can be pre-filled using an :term:`API` endpoint."""
+
+    class ServiceBindingPortInput(BaseModel):
+        edge_port: UUIDstr
+        geant_sid: str
+        is_tagged: bool
+        vlan_id: VLAN_ID
+        custom_firewall_filters: bool
+
+    class ImportLayer2CircuitForm(FormPage):
+        model_config = ConfigDict(title="Import Layer 2 Circuit")
+
+        service_type: Layer2CircuitServiceType
+        partner: str
+        layer_2_circuit_side_a: ServiceBindingPortInput
+        layer_2_circuit_side_b: ServiceBindingPortInput
+        virtual_circuit_id: VC_ID
+        layer_2_circuit_type: Layer2CircuitType
+        vlan_range_lower_bound: VLAN_ID | None = None
+        vlan_range_upper_bound: VLAN_ID | None = None
+        policer_enabled: bool = False
+        bandwidth: BandwidthString | None = None
+
+        @model_validator(mode="after")
+        def partner_id_matches_edge_port_owner(self) -> Self:
+            """Validate that the entered partner owns both Edge Ports."""
+            partner_id = get_partner_by_name(self.partner)["partner_id"]
+            if partner_id != self.layer_2_circuit_side_a.edge_port:
+                msg = f"Selected Edge Port on side A is not owned by partner {self.partner}."
+                raise ValueError(msg)
+            if partner_id != self.layer_2_circuit_side_b.edge_port:
+                msg = f"Selected Edge Port on side B is not owned by partner {self.partner}."
+                raise ValueError(msg)
+            return self
+
+        @model_validator(mode="after")
+        def tagged_layer_2_circuit_has_vlan_bounds(self) -> Self:
+            """If a Layer 2 Circuit is tagged, it must have a :term:`VLAN` range set."""
+            if self.layer_2_circuit_type == Layer2CircuitType.TAGGED and (
+                self.vlan_range_lower_bound is None or self.vlan_range_upper_bound is None
+            ):
+                msg = (
+                    f"A tagged Layer 2 Circuit must have a VLAN range set. Received lower: "
+                    f"{self.vlan_range_lower_bound}, upper: {self.vlan_range_upper_bound}."
+                )
+                raise ValueError(msg)
+            return self
+
+    user_input = yield ImportLayer2CircuitForm
+    return user_input.model_dump()
+
+
+@step("Create subscription")
+def create_subscription(partner: str, service_type: Layer2CircuitServiceType) -> State:
+    """Create a new subscription object."""
+    partner_id = get_partner_by_name(partner)["partner_id"]
+    product_id = get_product_id_by_name(ProductName(service_type))
+    subscription = ImportedLayer2Circuit.from_product_id(product_id, partner_id)
+
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+    subscription: ImportedLayer2CircuitInactive,
+    vlan_range_lower_bound: VLAN_ID,
+    vlan_range_upper_bound: VLAN_ID,
+) -> State:
+    """Initialize the subscription object."""
+    subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
+    subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
+    return {"subscription": subscription}

gso/workflows/l2_circuit/create_layer_2_circuit.py

+"""Workflow for creating a new Layer 2 Circuit."""
+
+from typing import Any, Self
+from uuid import uuid4
+
+from orchestrator import step, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import SubscriptionLifecycle
+from orchestrator.workflow import StepList, begin, done
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from orchestrator.workflows.utils import wrap_create_initial_input_form
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+from pydantic_forms.validators import Divider, Label, ReadOnlyField
+
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockInactive, Layer2CircuitType
+from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitInactive
+from gso.services.partners import get_partner_by_name
+from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id, partner_choice
+from gso.utils.shared_enums import SBPType
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.virtual_identifiers import VLAN_ID
+
+
+def initial_input_generator(product_name: str) -> FormGenerator:
+    """Gather input from the operator about a new Layer 2 Circuit subscription."""
+    geant_partner_id = get_partner_by_name("GEANT")["partner_id"]
+
+    class CreateLayer2CircuitServicePage(FormPage):
+        model_config = ConfigDict(title=f"{product_name}")
+
+        tt_number: TTNumber
+        partner: partner_choice() = geant_partner_id  # type: ignore[valid-type]
+        divider: Divider = Field(None, exclude=True)
+
+        layer_2_circuit_type: Layer2CircuitType
+        policer_enabled: bool = False
+
+    initial_user_input = yield CreateLayer2CircuitServicePage
+
+    class Layer2CircuitSideSelection(BaseModel):
+        edge_port: active_edge_port_selector(  # type: ignore[valid-type]
+            partner_id=initial_user_input.partner if initial_user_input.partner != geant_partner_id else None
+        )
+        vlan_id: VLAN_ID
+
+    def _vlan_range_field(*, is_tagged: bool) -> VLAN_ID:
+        """Return the appropriate field type based on whether the circuit is tagged."""
+        return VLAN_ID if is_tagged else ReadOnlyField(None, default_type=int)
+
+    def _policer_field(*, policer_enabled: bool) -> BandwidthString:
+        """Return the appropriate field type based on whether the policer is enabled."""
+        return BandwidthString if policer_enabled else ReadOnlyField(None, default_type=str)
+
+    class Layer2CircuitServiceSidesPage(FormPage):
+        model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports")
+
+        vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True)
+        vlan_range_lower_bound: _vlan_range_field(  # type: ignore[valid-type]
+            is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED
+        )
+        vlan_range_upper_bound: _vlan_range_field(  # type: ignore[valid-type]
+            is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED
+        )
+        vlan_divider: Divider = Field(None, exclude=True)
+        policer_bandwidth: _policer_field(policer_enabled=initial_user_input.policer_enabled)  # type: ignore[valid-type]
+        policer_burst_rate: _policer_field(policer_enabled=initial_user_input.policer_enabled)  # type: ignore[valid-type]
+        geant_sid: str
+        layer_2_circuit_side_a: Layer2CircuitSideSelection
+        side_divider: Divider = Field(None, exclude=True)
+        layer_2_circuit_side_b: Layer2CircuitSideSelection
+
+        @model_validator(mode="after")
+        def check_unique_sides(self) -> Self:
+            if self.layer_2_circuit_side_a.edge_port == self.layer_2_circuit_side_b.edge_port:
+                msg = "Both sides of the circuit cannot be connected to the same edge port"
+                raise ValueError(msg)
+            return self
+
+    layer_2_circuit_input = yield Layer2CircuitServiceSidesPage
+
+    return {"product_name": product_name} | initial_user_input.model_dump() | layer_2_circuit_input.model_dump()
+
+
+@step("Create subscription")
+def create_subscription(product: UUIDstr, partner: str) -> State:
+    """Create a new subscription object in the database."""
+    subscription = Layer2CircuitInactive.from_product_id(product, partner)
+
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+    subscription: Layer2CircuitInactive,
+    layer_2_circuit_side_a: dict[str, Any],
+    layer_2_circuit_side_b: dict[str, Any],
+    layer_2_circuit_type: Layer2CircuitType,
+    vlan_range_lower_bound: VLAN_ID | None,
+    vlan_range_upper_bound: VLAN_ID | None,
+    policer_enabled: bool,  # noqa: FBT001
+    policer_bandwidth: BandwidthString | None,
+    policer_burst_rate: BandwidthString | None,
+    geant_sid: str,
+) -> State:
+    """Build a subscription object from all user input."""
+    layer_2_circuit_sides = []
+    for circuit_side_data in [layer_2_circuit_side_a, layer_2_circuit_side_b]:
+        sbp = ServiceBindingPortInactive.new(
+            uuid4(),
+            edge_port=EdgePort.from_subscription(subscription_id=circuit_side_data["edge_port"]).edge_port,
+            sbp_type=SBPType.L2,
+            vlan_id=circuit_side_data["vlan_id"],
+            geant_sid=geant_sid,
+            is_tagged=layer_2_circuit_type == Layer2CircuitType.TAGGED,
+            custom_firewall_filters=False,
+        )
+        layer2_circuit_side = Layer2CircuitSideBlockInactive.new(uuid4(), sbp=sbp)
+        layer_2_circuit_sides.append(layer2_circuit_side)
+    subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
+    subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id()
+    subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
+    subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
+    subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
+    subscription.layer_2_circuit.policer_enabled = policer_enabled
+    subscription.layer_2_circuit.bandwidth = policer_bandwidth
+    subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate
+    subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"
+
+    subscription = Layer2Circuit.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Create Layer 2 Circuit Service",
+    initial_input_form=wrap_create_initial_input_form(initial_input_generator),
+    target=Target.CREATE,
+)
+def create_layer_2_circuit() -> StepList:
+    """Create a new Layer 2 Circuit service subscription."""
+    return (
+        begin
+        >> create_subscription
+        >> store_process_subscription(Target.CREATE)
+        >> initialize_subscription
+        >> set_status(SubscriptionLifecycle.ACTIVE)
+        >> resync
+        >> done
+    )

gso/workflows/l2_circuit/modify_layer_2_circuit.py

+"""A modification workflow for a Layer 2 Circuit subscription."""
+
+from orchestrator import begin, done, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, UUIDstr
+from orchestrator.workflow import StepList, step
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import ConfigDict, Field
+from pydantic_forms.validators import Divider, Label, ReadOnlyField
+
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.services.partners import get_partner_by_id
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.virtual_identifiers import VLAN_ID
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    """Get input from the operator about the modifications to be made to a Layer 2 Circuit subscription."""
+    subscription = Layer2Circuit.from_subscription(subscription_id)
+    product_name = subscription.product.name
+
+    class ModifyL2CircuitForm(FormPage):
+        model_config = ConfigDict(title=f"Modify {product_name}")
+        tt_number: TTNumber
+        partner: ReadOnlyField(get_partner_by_id(subscription.customer_id).name, default_type=str)  # type: ignore[valid-type]
+        divider: Divider = Field(None, exclude=True)
+
+        layer_2_circuit_type: Layer2CircuitType = subscription.layer_2_circuit.layer_2_circuit_type
+        policer_enabled: bool = subscription.layer_2_circuit.policer_enabled
+
+    layer_2_circuit_input = yield ModifyL2CircuitForm
+
+    class ModifyLayer2CircuitServiceSidesPage(FormPage):
+        model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports")
+
+        vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True)
+        if layer_2_circuit_input.layer_2_circuit_type == Layer2CircuitType.TAGGED:
+            vlan_range_lower_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_lower_bound  # type: ignore[assignment]
+            vlan_range_upper_bound: VLAN_ID = subscription.layer_2_circuit.vlan_range_upper_bound  # type: ignore[assignment]
+        else:
+            vlan_range_lower_bound: ReadOnlyField(None, default_type=int)  # type: ignore[no-redef, valid-type]
+            vlan_range_upper_bound: ReadOnlyField(None, default_type=int)  # type: ignore[no-redef, valid-type]
+
+        vlan_divider: Divider = Field(None, exclude=True)
+        if layer_2_circuit_input.policer_enabled:
+            policer_bandwidth: BandwidthString = subscription.layer_2_circuit.bandwidth  # type: ignore[assignment]
+            policer_burst_rate: BandwidthString = subscription.layer_2_circuit.policer_burst_rate  # type: ignore[assignment]
+        else:
+            policer_bandwidth: ReadOnlyField(None, default_type=str)  # type: ignore[no-redef, valid-type]
+            policer_burst_rate: ReadOnlyField(None, default_type=str)  # type: ignore[no-redef, valid-type]
+        policer_divider: Divider = Field(None, exclude=True)
+
+        layer_2_circuit_side_a: ReadOnlyField(  # type: ignore[valid-type]
+            EdgePort.from_subscription(
+                subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id
+            ).description,
+            default_type=str,
+        )
+        side_divider: Divider = Field(None, exclude=True)
+        layer_2_circuit_side_b: ReadOnlyField(  # type: ignore[valid-type]
+            EdgePort.from_subscription(
+                subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id
+            ).description,
+            default_type=str,
+        )
+
+    layer_2_circuit_sides = yield ModifyLayer2CircuitServiceSidesPage
+
+    return {"product_name": product_name} | layer_2_circuit_input.model_dump() | layer_2_circuit_sides.model_dump()
+
+
+@step("Update Layer 2 Circuit subscription")
+def modify_layer_2_circuit_subscription(
+    subscription: Layer2Circuit,
+    layer_2_circuit_type: Layer2CircuitType,
+    vlan_range_lower_bound: VLAN_ID | None,
+    vlan_range_upper_bound: VLAN_ID | None,
+    policer_enabled: bool,  # noqa: FBT001
+    policer_bandwidth: BandwidthString | None,
+    policer_burst_rate: BandwidthString | None,
+) -> dict:
+    """Update the Layer 2 Circuit subscription with the new values."""
+    subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
+    subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
+    subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
+    subscription.layer_2_circuit.policer_enabled = policer_enabled
+    subscription.layer_2_circuit.bandwidth = policer_bandwidth
+    subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate
+    for layer_2_circuit_side in subscription.layer_2_circuit.layer_2_circuit_sides:
+        layer_2_circuit_side.sbp.is_tagged = layer_2_circuit_type == Layer2CircuitType.TAGGED
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Modify Layer 2 Circuit Service",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def modify_layer_2_circuit() -> StepList:
+    """Modify an existing Layer 2 Circuit service subscription."""
+    return (
+        begin
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> modify_layer_2_circuit_subscription
+        >> resync
+        >> done
+    )

gso/workflows/l2_circuit/terminate_layer_2_circuit.py

+"""Workflow for terminating a Layer 2 Circuit."""
+
+from orchestrator import begin, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, done
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic_forms.types import FormGenerator
+
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.utils.types.tt_number import TTNumber
+
+
+def _input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    layer_2_circuit = Layer2Circuit.from_subscription(subscription_id)
+
+    class TerminateForm(FormPage):
+        tt_number: TTNumber
+
+    yield TerminateForm
+    return {"subscription": layer_2_circuit}
+
+
+@workflow(
+    "Terminate Layer 2 Circuit",
+    initial_input_form=wrap_modify_initial_input_form(_input_form_generator),
+    target=Target.TERMINATE,
+)
+def terminate_layer_2_circuit() -> StepList:
+    """Terminate a Layer 2 Circuit subscription."""
+    return (
+        begin
+        >> store_process_subscription(Target.TERMINATE)
+        >> unsync
+        >> set_status(SubscriptionLifecycle.TERMINATED)
+        >> resync
+        >> done
+    )