Skip to content

Create imported commercial peer

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

PeeringConnectionModel

Bases: BaseModel

Peering connection model for import Commercial Peer workflows.

Source code in gso/workflows/commercial_peer/create_imported_commercial_peer.py 
25
26
27
28
29
30
31
32
class PeeringConnectionModel(BaseModel):
    """Peering connection model for import Commercial Peer workflows."""

    bgp_session_v4: BGPPeerImportModelSimple
    bgp_session_v6: BGPPeerImportModelSimple
    minimum_hold_timer: int | None = None
    session_state: SessionState
    placement_port: UUIDstr

initial_input_form_generator()

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

Source code in gso/workflows/commercial_peer/create_imported_commercial_peer.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def initial_input_form_generator() -> FormGenerator:
    """Take all information passed to this workflow by the API endpoint that was called."""

    class ImportCommercialPeer(SubmitFormPage):
        partner: str
        peering_connection_list: list[PeeringConnectionModel] = Field(default_factory=list)
        prefix_limit_v4: NonNegativeInt | None = None
        prefix_limit_v6: NonNegativeInt | None = None
        partner_asn: NonNegativeInt | None = None
        partner_org_name: str | None = None

    user_input = yield ImportCommercialPeer

    return user_input.model_dump()

create_subscription(partner)

Create a new subscription object in the database.

Source code in gso/workflows/commercial_peer/create_imported_commercial_peer.py 
51
52
53
54
55
56
57
@step("Create subscription")
def create_subscription(partner: str) -> dict:
    """Create a new subscription object in the database."""
    partner_id = get_partner_by_name(partner).partner_id
    product_id = get_product_id_by_name(ProductName.IMPORTED_COMMERCIAL_PEER)
    subscription = ImportedCommercialPeerInactive.from_product_id(product_id, partner_id)
    return {"subscription": subscription, "subscription_id": subscription.subscription_id}

initialize_subscription(subscription, peering_connection_list, partner, prefix_limit_v4=None, prefix_limit_v6=None, partner_asn=None, partner_org_name=None)

Initialize the subscription with the user input.

Source code in gso/workflows/commercial_peer/create_imported_commercial_peer.py 
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("Initialize subscription")
def initialize_subscription(
    subscription: ImportedCommercialPeerInactive,
    peering_connection_list: list[dict[str, Any]],
    partner: str,
    prefix_limit_v4: NonNegativeInt | None = None,
    prefix_limit_v6: NonNegativeInt | None = None,
    partner_asn: NonNegativeInt | None = None,
    partner_org_name: str | None = None,
) -> dict:
    """Initialize the subscription with the user input."""
    subscription.commercial_peer.prefix_limit_v4 = prefix_limit_v4
    subscription.commercial_peer.prefix_limit_v6 = prefix_limit_v6
    subscription.commercial_peer.partner_asn = partner_asn
    subscription.commercial_peer.partner_org_name = partner_org_name
    subscription.commercial_peer.peering_connection_list = [
        PeeringConnectionInactive.new(
            subscription_id=uuid4(),
            bgp_session_v4=BGPSessionInactive.new(
                subscription_id=uuid4(),
                ip_type=IPTypes.IPV4,
                prefix_limit=prefix_limit_v4,
                **peering_connection["bgp_session_v4"],
            ),
            bgp_session_v6=BGPSessionInactive.new(
                subscription_id=uuid4(),
                ip_type=IPTypes.IPV6,
                prefix_limit=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.description = f"Commercial Peer - {partner}"
    return {"subscription": subscription}

create_imported_commercial_peer()

Create an Imported Commercial Peer without provisioning it.

Source code in gso/workflows/commercial_peer/create_imported_commercial_peer.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@workflow(
    "Create Imported Commercial Peer",
    initial_input_form=initial_input_form_generator,
    target=Target.CREATE,
)
def create_imported_commercial_peer() -> StepList:
    """Create an Imported Commercial Peer without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription()
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )