Skip to content

Import access ports l3 core service

A modification workflow for importing Access Ports to an L3 Core Service subscription.

initial_input_form_generator(subscription_id)

Get the initial input form for importing Access Ports to an L3 Core Service subscription.

Source code in gso/workflows/l3_core_service/import_access_ports_l3_core_service.py 
23
24
25
26
27
28
29
30
31
32
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Get the initial input form for importing Access Ports to an L3 Core Service subscription."""
    subscription = SubscriptionModel.from_subscription(subscription_id)

    class ImportAccessPortsForm(SubmitFormPage):
        service_binding_ports: list[ServiceBindingPort] = Field(default_factory=list)

    user_input = yield ImportAccessPortsForm

    return user_input.model_dump() | {"subscription": subscription}

add_access_ports_to_subscription(subscription, service_binding_ports)

Add Access Ports to the L3 Core Service subscription.

Source code in gso/workflows/l3_core_service/import_access_ports_l3_core_service.py 
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
92
93
94
95
96
97
98
@step("Modify subscription to add Access Ports")
def add_access_ports_to_subscription(
    subscription: SubscriptionModel,
    service_binding_ports: list[dict[str, Any]],
) -> dict:
    """Add Access Ports to the L3 Core Service subscription."""
    added_access_ports = []

    for sbp_data in service_binding_ports:
        # Create BGP sessions
        bgp_sessions = [
            BGPSession.new(
                subscription_id=uuid4(),
                ip_type=IPTypes.IPV4 if bgp_peer["peer_address"].version == IPV4_VERSION else IPTypes.IPV6,
                **bgp_peer,
            )
            for bgp_peer in sbp_data.get("bgp_peers", [])
        ]

        # Create BFD settings
        v4_bfd = BFDSettings.new(
            subscription_id=uuid4(),
            **sbp_data.get("v4_bfd_settings", {}),
        )
        v6_bfd = BFDSettings.new(
            subscription_id=uuid4(),
            **sbp_data.get("v6_bfd_settings", {}),
        )

        # Get the edge port
        edge_port = SubscriptionModel.from_subscription(sbp_data["edge_port"]).product_block  # type: ignore[attr-defined]

        # Create Service Binding Port
        sbp = ServiceBindingPortBlock.new(
            subscription_id=uuid4(),
            gs_id=sbp_data["gs_id"],
            sbp_type=sbp_data["sbp_type"],
            is_tagged=sbp_data.get("is_tagged", False),
            vlan_id=sbp_data["vlan_id"],
            custom_firewall_filters=sbp_data.get("custom_firewall_filters", False),
            ipv4_address=sbp_data.get("ipv4_address"),
            ipv4_mask=sbp_data.get("ipv4_mask"),
            ipv6_address=sbp_data.get("ipv6_address"),
            ipv6_mask=sbp_data.get("ipv6_mask"),
            bgp_sessions=bgp_sessions,
            v4_bfd_settings=v4_bfd,
            v6_bfd_settings=v6_bfd,
            edge_port=edge_port,
        )

        # Create Access Port
        access_port = AccessPort.new(
            subscription_id=uuid4(),
            ap_type=sbp_data["ap_type"],
            sbp=sbp,
            custom_service_name=sbp_data.get("custom_service_name"),
        )

        added_access_ports.append(access_port)

    # Add to the subscription's ap_list
    subscription.l3_core.ap_list.extend(added_access_ports)  # type: ignore[attr-defined]

    return {"subscription": subscription}

import_access_ports_l3_core_service()

Import Access Ports to an existing L3 Core Service subscription.

Source code in gso/workflows/l3_core_service/import_access_ports_l3_core_service.py 
101
102
103
104
@modify_workflow("Import Access Ports to L3 Core Service", initial_input_form=initial_input_form_generator)
def import_access_ports_l3_core_service() -> StepList:
    """Import Access Ports to an existing L3 Core Service subscription."""
    return begin >> add_access_ports_to_subscription >> resync