Skip to content

Create commercial peer

A creation workflow for adding a new Commercial Peer to the service database.

initial_input_form_generator(product_name)

Gather information to create a new Commercial Peer.

Source code in gso/workflows/commercial_peer/create_commercial_peer.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def initial_input_form_generator(product_name: str) -> FormGenerator:
    """Gather information to create a new Commercial Peer."""

    class CreateCommercialPeerForm(FormPage):
        model_config = ConfigDict(title=product_name)

        tt_number: TTNumber
        partner: partner_choice()  # type: ignore[valid-type]
        prefix_limit_v4: NonNegativeInt
        prefix_limit_v6: NonNegativeInt
        partner_asn: NonNegativeInt
        partner_org_name: str

        @model_validator(mode="after")
        def check_commercial_peer_with_same_partner(self) -> Self:
            """Ensure that the partner is not already used in a commercial peer subscription."""
            if get_active_commercial_peer_by_partner_id(self.partner):
                error_msg = (
                    f"A {product_name} subscription already exists for partner {self.partner}. "
                    "Please choose a different partner."
                )
                raise ValueError(error_msg)
            return self

    initial_user_input = yield CreateCommercialPeerForm

    return initial_user_input.model_dump()

create_subscription(product, partner)

Create a new subscription object in the service database.

Source code in gso/workflows/commercial_peer/create_commercial_peer.py 
49
50
51
52
53
54
55
56
57
@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object in the service database."""
    subscription = CommercialPeerInactive.from_product_id(product, partner)

    return {
        "subscription": subscription,
        "subscription_id": subscription.subscription_id,
    }

initialize_subscription(subscription, prefix_limit_v4, prefix_limit_v6, partner_asn, partner_org_name)

Initialise the subscription object with all user input.

Source code in gso/workflows/commercial_peer/create_commercial_peer.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@step("Initialize subscription")
def initialize_subscription(
    subscription: CommercialPeerInactive,
    prefix_limit_v4: NonNegativeInt,
    prefix_limit_v6: NonNegativeInt,
    partner_asn: NonNegativeInt,
    partner_org_name: str,
) -> State:
    """Initialise the subscription object with all user input."""
    partner_name = get_partner_by_id(subscription.customer_id).name
    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.description = f"Commercial Peer - {partner_name}"

    subscription = CommercialPeer.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)

    return {"subscription": subscription}

create_commercial_peer()

Create a new commercial peer without any Peering Connection.

Source code in gso/workflows/commercial_peer/create_commercial_peer.py 
81
82
83
84
@create_workflow("Create Commercial Peer", initial_input_form=initial_input_form_generator)
def create_commercial_peer() -> StepList:
    """Create a new commercial peer without any Peering Connection."""
    return begin >> create_subscription >> store_process_subscription() >> initialize_subscription