create_site.py

"""A creation workflow for adding a new site to the service database."""

from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.workflow import StepList, 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 pydantic_forms.core import ReadOnlyField

from gso.products.product_blocks import site as site_pb
from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
from gso.products.product_types import site
from gso.services.partners import get_partner_by_name
from gso.utils.helpers import BaseSiteValidatorModel


def initial_input_form_generator(product_name: str) -> FormGenerator:
    """Get input from the operator about the new site subscription."""

    class CreateSiteForm(FormPage, BaseSiteValidatorModel):
        class Config:
            title = product_name

        partner: str = ReadOnlyField("GEANT")
        site_name: str
        site_city: str
        site_country: str
        site_country_code: str
        site_latitude: LatitudeCoordinate
        site_longitude: LongitudeCoordinate
        site_bgp_community_id: int
        site_internal_id: int
        site_tier: site_pb.SiteTier
        site_ts_address: str

    user_input = yield CreateSiteForm

    return user_input.dict()


@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object in the service database."""
    subscription = site.SiteInactive.from_product_id(product, get_partner_by_name(partner)["id"])

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


@step("Initialize subscription")
def initialize_subscription(
    subscription: site.SiteInactive,
    site_name: str,
    site_city: str,
    site_country: str,
    site_country_code: str,
    site_latitude: LatitudeCoordinate,
    site_longitude: LongitudeCoordinate,
    site_bgp_community_id: int,
    site_internal_id: int,
    site_ts_address: str,
    site_tier: site_pb.SiteTier,
) -> State:
    """Initialise the subscription object with all user input."""
    subscription.site.site_name = site_name
    subscription.site.site_city = site_city
    subscription.site.site_country = site_country
    subscription.site.site_country_code = site_country_code
    subscription.site.site_latitude = site_latitude
    subscription.site.site_longitude = site_longitude
    subscription.site.site_bgp_community_id = site_bgp_community_id
    subscription.site.site_internal_id = site_internal_id
    subscription.site.site_tier = site_tier
    subscription.site.site_ts_address = site_ts_address

    subscription.description = f"Site in {site_city}, {site_country}"

    subscription = site.SiteProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)

    return {"subscription": subscription}


@workflow(
    "Create Site",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_site() -> StepList:
    """Create a new site subscription."""
    return (
        init
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )