Skip to content

Base create imported placement port

Base functionality for creating an imported placement port subscription.

L3Interface

Bases: BaseModel

L3 Interface model for imported placement port subscriptions.

Source code in gso/workflows/placement_port/base_create_imported_placement_port.py 
20
21
22
23
24
25
26
27
28
29
30
31
class L3Interface(BaseModel):
    """L3 Interface model for imported placement port subscriptions."""

    gs_id: IMPORTED_GS_ID | None = None
    edge_port: str
    is_tagged: bool = True
    vlan_id: VLAN_ID | None = None
    ipv4_address: IPv4AddressType | None = None
    ipv4_mask: IPv4Netmask | None = None
    ipv6_address: IPv6AddressType | None = None
    ipv6_mask: IPv6Netmask | None = None
    custom_firewall_filters: bool = False

initial_input_generator()

Take all information passed to this workflow by the API endpoint that was called.

Source code in gso/workflows/placement_port/base_create_imported_placement_port.py 
34
35
36
37
38
39
40
41
42
43
44
def initial_input_generator() -> FormGenerator:
    """Take all information passed to this workflow by the API endpoint that was called."""

    class ImportPlacementPortForm(FormPage):
        l3_interface: L3Interface
        product_name: PlacementPortNameType
        partner: str

    user_input = yield ImportPlacementPortForm

    return user_input.model_dump()

initialize_subscription(subscription, l3_interface)

Initialize the placement port subscription with the provided parameters.

Source code in gso/workflows/placement_port/base_create_imported_placement_port.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@step("Initialize subscription")
def initialize_subscription(
    subscription: SubscriptionModel,
    l3_interface: dict[str, Any],
) -> State:
    """Initialize the placement port subscription with the provided parameters."""
    edge_port_subscription = EdgePort.from_subscription(l3_interface.pop("edge_port"))
    subscription.l3_interface = subscription.l3_interface.new(  # type: ignore[attr-defined]
        uuid4(),
        edge_port=edge_port_subscription.edge_port,
        **l3_interface,
    )

    subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
    subscription.description = f"{subscription.product.name} - {l3_interface['gs_id']}"

    return {"subscription": subscription}