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 )