Skip to content

Create imported site

A creation workflow for importing an existing site.

create_subscription(partner)

Create a new subscription object in the service database.

Source code in gso/workflows/site/create_imported_site.py 
23
24
25
26
27
28
29
30
31
32
33
@step("Create subscription")
def create_subscription(partner: str) -> State:
    """Create a new subscription object in the service database."""
    partner_id = get_partner_by_name(partner).partner_id
    product_id: UUID = subscriptions.get_product_id_by_name(ProductName.IMPORTED_SITE)
    subscription = ImportedSiteInactive.from_product_id(product_id, partner_id)

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

generate_initial_input_form()

Generate a form that is filled in using information passed through the API endpoint.

Source code in gso/workflows/site/create_imported_site.py 
36
37
38
39
40
41
42
43
def generate_initial_input_form() -> FormGenerator:
    """Generate a form that is filled in using information passed through the API endpoint."""

    class ImportSite(FormPage, BaseSiteValidatorModel):
        model_config = ConfigDict(title="Import Site")

    user_input = yield ImportSite
    return user_input.model_dump()

initialize_subscription(subscription, site_name, site_city, site_country, site_country_code, site_latitude, site_longitude, site_bgp_community_id, site_internal_id, site_ts_address, site_tier, site_contains_optical_equipment)

Initialise the subscription object with all input.

Source code in gso/workflows/site/create_imported_site.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@step("Initialize subscription")
def initialize_subscription(
    subscription: ImportedSiteInactive,
    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: IPAddress,
    site_tier: SiteTier,
    site_contains_optical_equipment: bool,  # noqa: FBT001
) -> State:
    """Initialise the subscription object with all 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.site.site_contains_optical_equipment = site_contains_optical_equipment

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

    return {"subscription": subscription}

create_imported_site()

Workflow to import a site without provisioning it.

Source code in gso/workflows/site/create_imported_site.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@workflow(
    "Import Site",
    target=Target.CREATE,
    initial_input_form=generate_initial_input_form,
)
def create_imported_site() -> StepList:
    """Workflow to import a site without provisioning it."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )