create_trunk_config.py

from typing import List, Optional
from uuid import uuid4
from random import randint

from orchestrator.db.models import ProductTable, SubscriptionTable
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, choice_list
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import done, init, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form

from products.product_types.trunk_config import TrunkConfigInactive, TrunkConfigProvisioning
from products.product_types.trunk import Trunk


def trunk_selector() -> list:
    trunk_subscriptions = {}
    for trunk_id, trunk_description in (
        SubscriptionTable.query.join(ProductTable)
        .filter(
            ProductTable.product_type == "Trunk",
            SubscriptionTable.status == "active",
        )
        .with_entities(SubscriptionTable.subscription_id, SubscriptionTable.description)
        .all()
    ):
        trunk_subscriptions[str(trunk_id)] = trunk_description

    return choice_list(
        Choice("TrunkEnum", zip(trunk_subscriptions.keys(), trunk_subscriptions.items())),
        min_items=1,
        max_items=1,
    )


def initial_input_form_generator(product_name: str) -> FormGenerator:
    class CreateTrunkConfigForm(FormPage):
        class Config:
            title = product_name

        trunk_ids: trunk_selector()

    user_input = yield CreateTrunkConfigForm

    return user_input.dict()


def _provision_in_group_management_system() -> int:

    return random.randint(0, 2147483648)


@step("Create subscription")
def create_subscription(product: UUIDstr) -> State:
    subscription = TrunkConfigInactive.from_product_id(product, uuid4())

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


@step("Initialize subscription")
def initialize_subscription(subscription: TrunkConfigInactive, trunk_ids: List[str]) -> State:
    subscription.trunk_config.trunk = Trunk.from_subscription(trunk_ids[0]).trunk
    subscription.description = f"TrunkConfig from Trunk {subscription.trunk_config.trunk.trunk_name}"
    subscription = TrunkConfigProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)

    return {"subscription": subscription}


@step("Provision trunk config")
def provision_trunk_config(subscription: TrunkConfigProvisioning) -> State:
    trunk_config_id = _provision_in_group_management_system()
    subscription.trunk_config.trunk_config_id = trunk_config_id

    return {"subscription": subscription}


@workflow(
    "Create trunk config",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_trunk():

    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> provision_trunk_config
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )