Skip to content

Modify layer 2 circuit

A modification workflow for a Layer 2 Circuit subscription.

initial_input_form_generator(subscription_id)

Get input from the operator about the modifications to be made to a Layer 2 Circuit subscription.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py
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
        custom_service_name: str | None = subscription.layer_2_circuit.custom_service_name

    layer_2_circuit_input = yield ModifyL2CircuitForm

    class ModifyLayer2CircuitServiceSidesPage(SubmitFormPage):
        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()

modify_layer_2_circuit_subscription(subscription, layer_2_circuit_type, vlan_range_lower_bound, vlan_range_upper_bound, policer_enabled, policer_bandwidth, policer_burst_rate, custom_service_name)

Update the Layer 2 Circuit subscription with the new values.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py
@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,
    custom_service_name: str | 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
    subscription.layer_2_circuit.custom_service_name = custom_service_name
    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}

modify_layer_2_circuit()

Modify an existing Layer 2 Circuit service subscription.

Source code in gso/workflows/l2_circuit/modify_layer_2_circuit.py
@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
    )