Skip to content

Create imported layer 2 circuit

A creation workflow that adds an existing Layer 2 Circuit to the database.

initial_input_form_generator()

Generate a form that can be pre-filled using an API endpoint.

Source code in gso/workflows/l2_circuit/create_imported_layer_2_circuit.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def initial_input_form_generator() -> FormGenerator:
    """Generate a form that can be pre-filled using an API endpoint."""

    class ServiceBindingPortInput(BaseModel):
        edge_port: UUIDstr
        vlan_id: VLAN_ID

    class ImportLayer2CircuitForm(SubmitFormPage):
        model_config = ConfigDict(title="Import Layer 2 Circuit")

        service_type: Layer2CircuitServiceType
        partner: str
        gs_id: IMPORTED_GS_ID
        vc_id: VC_ID
        layer_2_circuit_side_a: ServiceBindingPortInput
        layer_2_circuit_side_b: ServiceBindingPortInput
        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
        policer_bandwidth: BandwidthString | None = None
        policer_burst_rate: BandwidthString | None = None
        custom_service_name: str | None = None

        @model_validator(mode="after")
        def vlan_layer_2_circuit_has_vlan_bounds(self) -> Self:
            """If a Layer 2 Circuit is VLAN, it must have a VLAN range set."""
            if self.layer_2_circuit_type == Layer2CircuitType.VLAN and (
                self.vlan_range_lower_bound is None or self.vlan_range_upper_bound is None
            ):
                msg = (
                    f"A VLAN 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

        @model_validator(mode="after")
        def policer_bandwidth_required_if_policer_enabled(self) -> Self:
            """If the policer is enabled, the bandwidth and burst rate must be set."""
            if self.policer_enabled and (self.policer_bandwidth is None or self.policer_burst_rate is None):
                msg = (
                    f"If the policer is enabled, the bandwidth and burst rate must be set. Received bandwidth: "
                    f"{self.policer_bandwidth}, burst rate: {self.policer_burst_rate}."
                )
                raise ValueError(msg)
            return self

    user_input = yield ImportLayer2CircuitForm
    return user_input.model_dump()

create_subscription(partner, service_type)

Create a new subscription object.

Source code in gso/workflows/l2_circuit/create_imported_layer_2_circuit.py 
83
84
85
86
87
88
89
90
@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 = ImportedLayer2CircuitInactive.from_product_id(product_id, partner_id)

    return {"subscription": subscription, "subscription_id": subscription.subscription_id}

initialize_subscription(subscription, gs_id, layer_2_circuit_side_a, layer_2_circuit_side_b, layer_2_circuit_type, vlan_range_lower_bound, vlan_range_upper_bound, policer_enabled, policer_bandwidth, policer_burst_rate, custom_service_name, vc_id)

Initialize the subscription object.

Source code in gso/workflows/l2_circuit/create_imported_layer_2_circuit.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@step("Initialize subscription")
def initialize_subscription(
    subscription: ImportedLayer2CircuitInactive,
    gs_id: str,
    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,
    custom_service_name: str | None,
    vc_id: VC_ID,
) -> State:
    """Initialize the subscription object."""
    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"],
            gs_id=gs_id,
            is_tagged=layer_2_circuit_type == Layer2CircuitType.VLAN,
            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 = 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.layer_2_circuit.custom_service_name = custom_service_name
    subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"

    return {"subscription": subscription}

create_imported_layer_2_circuit()

Import a Layer 2 Circuit without provisioning it.

Source code in gso/workflows/l2_circuit/create_imported_layer_2_circuit.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@workflow(
    "Create imported Layer 2 Circuit",
    initial_input_form=initial_input_form_generator,
    target=Target.CREATE,
)
def create_imported_layer_2_circuit() -> StepList:
    """Import a Layer 2 Circuit without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )