Skip to content

Create imported edge port

A creation workflow that adds an existing Edge Port to the DB.

create_subscription(partner)

Create a new subscription object.

Source code in gso/workflows/edge_port/create_imported_edge_port.py 
27
28
29
30
31
32
33
34
@step("Create subscription")
def create_subscription(partner: str) -> State:
    """Create a new subscription object."""
    partner_id = get_partner_by_name(partner).partner_id
    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}

initial_input_form_generator()

Generate a form that is filled in using information passed through the API endpoint.

Source code in gso/workflows/edge_port/create_imported_edge_port.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def initial_input_form_generator() -> FormGenerator:
    """Generate a form that is filled in using information passed through the API endpoint."""

    class ImportEdgePort(SubmitFormPage):
        model_config = ConfigDict(title="Import Edge Port")

        node: active_pe_router_selector()  # type: ignore[valid-type]
        partner: str
        service_type: EdgePortType
        enable_lacp: bool
        speed: PhysicalPortCapacity
        encapsulation: EncapsulationType = EncapsulationType.DOT1Q
        minimum_links: int
        mac_address: str | None = None
        ignore_if_down: bool = False
        ga_id: IMPORTED_GA_ID | None = None
        description: str | None = None
        name: str
        ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]
        custom_service_name: str | None = None

    user_input = yield ImportEdgePort

    return user_input.model_dump()

initialize_subscription(subscription, node, service_type, speed, encapsulation, name, minimum_links, ga_id, mac_address, partner, enable_lacp, ignore_if_down, ae_members, description=None, custom_service_name=None)

Initialise the subscription object in the service database.

Source code in gso/workflows/edge_port/create_imported_edge_port.py 
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@step("Initialize subscription")
def initialize_subscription(
    subscription: ImportedEdgePortInactive,
    node: UUIDstr,
    service_type: EdgePortType,
    speed: PhysicalPortCapacity,
    encapsulation: EncapsulationType,
    name: str,
    minimum_links: int,
    ga_id: IMPORTED_GA_ID | None,
    mac_address: str | None,
    partner: str,
    enable_lacp: bool,  # noqa: FBT001
    ignore_if_down: bool,  # noqa: FBT001
    ae_members: list[dict[str, Any]],
    description: str | None = None,
    custom_service_name: str | None = None,
) -> State:
    """Initialise the subscription object in the service database."""
    router = Router.from_subscription(node).router
    subscription.edge_port.node = router
    subscription.edge_port.edge_port_name = name
    subscription.edge_port.edge_port_description = description
    subscription.edge_port.enable_lacp = enable_lacp
    subscription.edge_port.encapsulation = encapsulation
    subscription.edge_port.mac_address = mac_address
    subscription.edge_port.member_speed = speed
    subscription.edge_port.minimum_links = minimum_links
    subscription.edge_port.edge_port_type = service_type
    subscription.edge_port.ignore_if_down = ignore_if_down
    subscription.edge_port.ga_id = ga_id
    subscription.edge_port.custom_service_name = custom_service_name
    subscription.description = f"Edge Port {name} on {router.router_fqdn}, {partner}, {ga_id or ""}"
    for member in ae_members:
        subscription.edge_port.edge_port_ae_members.append(
            EdgePortAEMemberBlockInactive.new(subscription_id=uuid4(), **member)
        )

    return {"subscription": subscription}

create_imported_edge_port()

Import an Edge Port without provisioning it.

Source code in gso/workflows/edge_port/create_imported_edge_port.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@workflow(
    "Import Edge Port",
    initial_input_form=initial_input_form_generator,
    target=Target.CREATE,
)
def create_imported_edge_port() -> StepList:
    """Import an Edge Port without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )