Skip to content

Import peering connections

A modification workflow for importing peering connections to a Commercial Peer subscription.

initial_input_form_generator(subscription_id)

Get the initial input form for importing peering connections to a Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/import_peering_connections.py 
20
21
22
23
24
25
26
27
28
29
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Get the initial input form for importing peering connections to a Commercial Peer subscription."""
    subscription = CommercialPeer.from_subscription(subscription_id)

    class ImportPeeringConnectionsForm(SubmitFormPage):
        peering_connection_list: list[PeeringConnectionModel] = Field(default_factory=list)

    user_input = yield ImportPeeringConnectionsForm

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

add_peering_connections_to_subscription(subscription, peering_connection_list)

Add peering connections to the Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/import_peering_connections.py 
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
@step("Modify subscription to add peering connections")
def add_peering_connections_to_subscription(
    subscription: CommercialPeer,
    peering_connection_list: list[dict[str, Any]],
) -> dict:
    """Add peering connections to the Commercial Peer subscription."""
    added_peering_connections = [
        PeeringConnection.new(
            subscription_id=uuid4(),
            bgp_session_v4=BGPSession.new(
                subscription_id=uuid4(),
                ip_type=IPTypes.IPV4,
                prefix_limit=subscription.commercial_peer.prefix_limit_v4,
                **peering_connection["bgp_session_v4"],
            ),
            bgp_session_v6=BGPSession.new(
                subscription_id=uuid4(),
                ip_type=IPTypes.IPV6,
                prefix_limit=subscription.commercial_peer.prefix_limit_v6,
                **peering_connection["bgp_session_v6"],
            ),
            minimum_hold_timer=peering_connection.get("minimum_hold_timer", None),
            session_state=peering_connection["session_state"],
            placement_port=SubscriptionModel.from_subscription(peering_connection["placement_port"]).product_block,  # type: ignore[attr-defined]
        )
        for peering_connection in peering_connection_list
    ]
    subscription.commercial_peer.peering_connection_list.extend(added_peering_connections)
    return {"subscription": subscription}

import_peering_connections()

Import Peering connections to an existing Commercial Peer subscription.

Source code in gso/workflows/commercial_peer/import_peering_connections.py 
63
64
65
66
@modify_workflow("Import Peering Connections", initial_input_form_generator)
def import_peering_connections() -> StepList:
    """Import Peering connections to an existing Commercial Peer subscription."""
    return begin >> add_peering_connections_to_subscription >> resync >> done