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
81
82
83
84
85
86
87
88
89
90
91
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
        partner_id: str
        is_primary: bool

    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

        # TODO: Uncomment when is_primary is added back to the form and make all is_primary required in all access
        # @model_validator(mode="after")
        # def only_one_side_can_be_primary(self) -> Self:
        #     """Ensure that both sides do not have is_primary set to True."""
        #     if self.layer_2_circuit_side_a.is_primary and self.layer_2_circuit_side_b.is_primary:
        #         msg = "Only one side of the Layer 2 Circuit can be marked as primary."  # noqa: ERA001
        #         raise ValueError(msg)  # noqa: ERA001
        #     return self  # noqa: ERA001

    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 
 94
 95
 96
 97
 98
 99
100
101
@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 
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@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,
            partner_id=circuit_side_data["partner_id"],
            is_primary=circuit_side_data["is_primary"],
        )
        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
    partner_a = get_partner_by_id(layer_2_circuit_side_a["partner_id"]).name
    partner_b = get_partner_by_id(layer_2_circuit_side_b["partner_id"]).name
    subscription.description = (
        f"{subscription.product.name} - {subscription.layer_2_circuit.custom_service_name} ({partner_a} - {partner_b})"
    )

    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 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@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()
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )