Skip to content

Create imported iptrunk

A creation workflow for adding an existing IP trunk to the service database.

initial_input_form_generator()

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

Source code in gso/workflows/iptrunk/create_imported_iptrunk.py 
27
28
29
30
31
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
def initial_input_form_generator() -> FormGenerator:
    """Take all information passed to this workflow by the API endpoint that was called."""

    class CreateIptrunkForm(SubmitFormPage):
        model_config = ConfigDict(title="Import Iptrunk")

        partner: str
        gs_id: IMPORTED_GS_ID | None = None
        iptrunk_description: str | None = None
        iptrunk_type: IptrunkType
        iptrunk_speed: PhysicalPortCapacity
        iptrunk_minimum_links: int
        iptrunk_isis_metric: int

        side_a_node_id: active_router_selector()  # type: ignore[valid-type]
        side_a_ae_iface: str
        side_a_ga_id: IMPORTED_GA_ID | None = None
        side_a_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]

        side_b_node_id: active_router_selector()  # type: ignore[valid-type]
        side_b_ae_iface: str
        side_b_ga_id: IMPORTED_GA_ID | None = None
        side_b_ae_members: Annotated[list[LAGMember], AfterValidator(validate_unique_list)]

        iptrunk_ipv4_network: ipaddress.IPv4Network
        iptrunk_ipv6_network: ipaddress.IPv6Network

    initial_user_input = yield CreateIptrunkForm

    return initial_user_input.model_dump()

create_subscription(partner)

Create a new subscription in the service database.

Source code in gso/workflows/iptrunk/create_imported_iptrunk.py 
59
60
61
62
63
64
65
66
67
68
69
@step("Create a new subscription")
def create_subscription(partner: str) -> State:
    """Create a new subscription in the service database."""
    partner_id = get_partner_by_name(partner).partner_id
    product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IP_TRUNK)
    subscription = ImportedIptrunkInactive.from_product_id(product_id, partner_id)

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

initialize_subscription(subscription, gs_id, iptrunk_type, iptrunk_description, iptrunk_speed, iptrunk_minimum_links, iptrunk_isis_metric, side_a_node_id, side_a_ae_iface, side_a_ga_id, side_a_ae_members, side_b_node_id, side_b_ae_iface, side_b_ga_id, side_b_ae_members)

Take all input from the user, and store it in the database.

Source code in gso/workflows/iptrunk/create_imported_iptrunk.py 
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@step("Initialize subscription")
def initialize_subscription(
    subscription: ImportedIptrunkInactive,
    gs_id: IMPORTED_GS_ID | None,
    iptrunk_type: IptrunkType,
    iptrunk_description: str,
    iptrunk_speed: PhysicalPortCapacity,
    iptrunk_minimum_links: int,
    iptrunk_isis_metric: int,
    side_a_node_id: str,
    side_a_ae_iface: str,
    side_a_ga_id: IMPORTED_GA_ID | None,
    side_a_ae_members: LAGMemberList,
    side_b_node_id: str,
    side_b_ae_iface: str,
    side_b_ga_id: IMPORTED_GA_ID | None,
    side_b_ae_members: LAGMemberList,
) -> State:
    """Take all input from the user, and store it in the database."""
    subscription.iptrunk.gs_id = gs_id
    subscription.iptrunk.iptrunk_description = iptrunk_description
    subscription.iptrunk.iptrunk_type = iptrunk_type
    subscription.iptrunk.iptrunk_speed = iptrunk_speed
    subscription.iptrunk.iptrunk_isis_metric = iptrunk_isis_metric
    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links

    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node = Router.from_subscription(side_a_node_id).router
    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface = side_a_ae_iface
    subscription.iptrunk.iptrunk_sides[0].ga_id = side_a_ga_id
    for member in side_a_ae_members:
        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members.append(
            IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
        )

    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node = Router.from_subscription(side_b_node_id).router
    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface = side_b_ae_iface
    subscription.iptrunk.iptrunk_sides[1].ga_id = side_b_ga_id
    for member in side_b_ae_members:
        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members.append(
            IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
        )
    side_names = sorted([
        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
    ])
    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"
    return {"subscription": subscription}

update_ipam_stub_for_subscription(subscription, iptrunk_ipv4_network, iptrunk_ipv6_network)

Update IPAM information in the subscription.

Source code in gso/workflows/iptrunk/create_imported_iptrunk.py 
121
122
123
124
125
126
127
128
129
130
131
@step("Update IPAM Stub for Subscription")
def update_ipam_stub_for_subscription(
    subscription: ImportedIptrunkInactive,
    iptrunk_ipv4_network: ipaddress.IPv4Network,
    iptrunk_ipv6_network: ipaddress.IPv6Network,
) -> State:
    """Update IPAM information in the subscription."""
    subscription.iptrunk.iptrunk_ipv4_network = iptrunk_ipv4_network
    subscription.iptrunk.iptrunk_ipv6_network = iptrunk_ipv6_network

    return {"subscription": subscription}

create_imported_iptrunk()

Import an IP trunk without provisioning it.

Source code in gso/workflows/iptrunk/create_imported_iptrunk.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@workflow(
    "Import iptrunk",
    initial_input_form=initial_input_form_generator,
    target=Target.CREATE,
)
def create_imported_iptrunk() -> StepList:
    """Import an IP trunk without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> update_ipam_stub_for_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )