Imports

CLI commands for importing data to coreDB.

CreatePartner

Bases: BaseModel

Required inputs for creating a partner.

Source code in gso/cli/imports.py

class CreatePartner(BaseModel):
    """Required inputs for creating a partner."""

    name: PartnerName
    email: PartnerEmail

SiteImportModel

Bases: BaseSiteValidatorModel

The required input for importing an existing gso.products.product_types.site.

Source code in gso/cli/imports.py

class SiteImportModel(BaseSiteValidatorModel):
    """The required input for importing an existing `gso.products.product_types.site`."""

RouterImportModel

Bases: BaseModel

Required fields for importing an existing gso.product.product_types.router.

Source code in gso/cli/imports.py

class RouterImportModel(BaseModel):
    """Required fields for importing an existing `gso.product.product_types.router`."""

    partner: str
    router_site: str
    hostname: str
    ts_port: int
    router_vendor: Vendor
    router_role: RouterRole
    router_lo_ipv4_address: IPv4AddressType
    router_lo_ipv6_address: IPv6AddressType
    router_lo_iso_address: str

SuperPopSwitchImportModel

Bases: BaseModel

Required fields for importing an existing gso.product.product_types.super_pop_switch.

Source code in gso/cli/imports.py

class SuperPopSwitchImportModel(BaseModel):
    """Required fields for importing an existing `gso.product.product_types.super_pop_switch`."""

    partner: str
    super_pop_switch_site: str
    hostname: str
    super_pop_switch_ts_port: PortNumber
    super_pop_switch_mgmt_ipv4_address: ipaddress.IPv4Address

OfficeRouterImportModel

Bases: BaseModel

Required fields for importing an existing gso.product.product_types.office_router.

Source code in gso/cli/imports.py

class OfficeRouterImportModel(BaseModel):
    """Required fields for importing an existing `gso.product.product_types.office_router`."""

    partner: str
    office_router_site: str
    office_router_fqdn: str
    office_router_ts_port: PortNumber
    office_router_lo_ipv4_address: ipaddress.IPv4Address
    office_router_lo_ipv6_address: ipaddress.IPv6Address

IptrunkImportModel

Bases: BaseModel

Required fields for importing an existing gso.products.product_types.iptrunk.

Source code in gso/cli/imports.py

class IptrunkImportModel(BaseModel):
    """Required fields for importing an existing `gso.products.product_types.iptrunk`."""

    partner: ExistingPartnerStr
    gs_id: IMPORTED_GS_ID | None
    iptrunk_type: IptrunkType
    iptrunk_description: str | None = None
    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[LAGMember]
    side_b_node_id: str
    side_b_ae_iface: str
    side_b_ga_id: IMPORTED_GA_ID | None
    side_b_ae_members: LAGMemberList[LAGMember]

    iptrunk_ipv4_network: ipaddress.IPv4Network
    iptrunk_ipv6_network: ipaddress.IPv6Network

    @classmethod
    def _get_active_routers(cls) -> set[str]:
        return {
            str(router["subscription_id"]) for router in get_active_router_subscriptions(includes=["subscription_id"])
        }

    @field_validator("side_a_node_id", "side_b_node_id")
    def check_if_router_side_is_available(cls, value: str) -> str:
        """Both sides of the trunk must exist in GSO."""
        if value not in cls._get_active_routers():
            msg = f"Router {value} not found"
            raise ValueError(msg)

        return value

    @model_validator(mode="after")
    def check_members(self) -> Self:
        """Amount of LAG members has to match on side A and B, and meet the minimum requirement."""
        len_a = len(self.side_a_ae_members)
        len_b = len(self.side_b_ae_members)

        if len_a < self.iptrunk_minimum_links:
            msg = f"Side A members should be at least {self.iptrunk_minimum_links} (iptrunk_minimum_links)"
            raise ValueError(msg)

        if len_a != len_b:
            msg = "Mismatch between Side A and B members"
            raise ValueError(msg)

        return self

check_if_router_side_is_available(value)

Both sides of the trunk must exist in GSO.

Source code in gso/cli/imports.py

@field_validator("side_a_node_id", "side_b_node_id")
def check_if_router_side_is_available(cls, value: str) -> str:
    """Both sides of the trunk must exist in GSO."""
    if value not in cls._get_active_routers():
        msg = f"Router {value} not found"
        raise ValueError(msg)

    return value

check_members()

Amount of LAG members has to match on side A and B, and meet the minimum requirement.

Source code in gso/cli/imports.py

@model_validator(mode="after")
def check_members(self) -> Self:
    """Amount of LAG members has to match on side A and B, and meet the minimum requirement."""
    len_a = len(self.side_a_ae_members)
    len_b = len(self.side_b_ae_members)

    if len_a < self.iptrunk_minimum_links:
        msg = f"Side A members should be at least {self.iptrunk_minimum_links} (iptrunk_minimum_links)"
        raise ValueError(msg)

    if len_a != len_b:
        msg = "Mismatch between Side A and B members"
        raise ValueError(msg)

    return self

OpenGearImportModel

Bases: BaseModel

Required fields for importing an existing gso.products.product_types.opengear.

Source code in gso/cli/imports.py

class OpenGearImportModel(BaseModel):
    """Required fields for importing an existing `gso.products.product_types.opengear`."""

    partner: str
    opengear_site: str
    opengear_hostname: str
    opengear_wan_address: IPv4AddressType
    opengear_wan_netmask: IPv4AddressType
    opengear_wan_gateway: IPv4AddressType

BaseBGPPeerImportModel

Bases: BaseModel

Base BGP Peer import model.

Source code in gso/cli/imports.py

class BaseBGPPeerImportModel(BaseModel):
    """Base BGP Peer import model."""

    bfd_enabled: bool = False
    has_custom_policies: bool = False
    authentication_key: str | None
    multipath_enabled: bool = False
    send_default_route: bool = False
    is_passive: bool = False
    peer_address: IPAddress
    families: list[IPFamily]
    is_multi_hop: bool
    rtbh_enabled: bool  # whether Remote Triggered Blackhole is enabled
    prefix_limit: NonNegativeInt | None = None
    ttl_security: NonNegativeInt | None = None

L3CoreServiceImportModel

Bases: BaseModel

Import L3 Core Service model.

Source code in gso/cli/imports.py

class L3CoreServiceImportModel(BaseModel):
    """Import L3 Core Service model."""

    class BFDSettingsModel(BaseModel):
        """BFD Settings model."""

        bfd_enabled: bool = False
        bfd_interval_rx: int | None = None
        bfd_interval_tx: int | None = None
        bfd_multiplier: int | None = None

    class ServiceBindingPort(BaseModel):
        """Service Binding model."""

        edge_port: Annotated[str, AfterValidator(validate_edge_port_exists)]
        ap_type: str
        custom_service_name: str | None = None
        gs_id: IMPORTED_GS_ID
        sbp_type: SBPType = SBPType.L3
        is_tagged: bool = False
        vlan_id: VLAN_ID
        custom_firewall_filters: bool = False
        ipv4_address: IPv4AddressType
        ipv4_mask: IPv4Netmask
        ipv6_address: IPv6AddressType
        ipv6_mask: IPv6Netmask
        is_multi_hop: bool = True
        bgp_peers: list[BaseBGPPeerImportModel]
        v4_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
        v6_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"

    partner: ExistingPartnerStr
    service_binding_ports: list[ServiceBindingPort]
    product_name: L3ProductNameType

BFDSettingsModel

Bases: BaseModel

BFD Settings model.

Source code in gso/cli/imports.py

class BFDSettingsModel(BaseModel):
    """BFD Settings model."""

    bfd_enabled: bool = False
    bfd_interval_rx: int | None = None
    bfd_interval_tx: int | None = None
    bfd_multiplier: int | None = None

ServiceBindingPort

Bases: BaseModel

Service Binding model.

Source code in gso/cli/imports.py

class ServiceBindingPort(BaseModel):
    """Service Binding model."""

    edge_port: Annotated[str, AfterValidator(validate_edge_port_exists)]
    ap_type: str
    custom_service_name: str | None = None
    gs_id: IMPORTED_GS_ID
    sbp_type: SBPType = SBPType.L3
    is_tagged: bool = False
    vlan_id: VLAN_ID
    custom_firewall_filters: bool = False
    ipv4_address: IPv4AddressType
    ipv4_mask: IPv4Netmask
    ipv6_address: IPv6AddressType
    ipv6_mask: IPv6Netmask
    is_multi_hop: bool = True
    bgp_peers: list[BaseBGPPeerImportModel]
    v4_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
    v6_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"

IASImportModel

Bases: L3CoreServiceImportModel

Import IAS model.

Source code in gso/cli/imports.py

class IASImportModel(L3CoreServiceImportModel):
    """Import IAS model."""

    ias_flavor: IASFlavor = IASFlavor.IAS_PS_OPT_OUT

RAndEImportModel

Bases: L3CoreServiceImportModel

Import R&E Peer and R&E LHCONE model.

Source code in gso/cli/imports.py

class RAndEImportModel(L3CoreServiceImportModel):
    """Import R&E Peer and R&E LHCONE model."""

    v4_bgp_local_preference: NonNegativeInt
    v4_bgp_med: MultiExitDiscriminator
    v6_bgp_local_preference: NonNegativeInt
    v6_bgp_med: MultiExitDiscriminator

Layer2CircuitServiceImportModel

Bases: BaseModel

Import Layer 2 Circuit Service model.

Source code in gso/cli/imports.py

class Layer2CircuitServiceImportModel(BaseModel):
    """Import Layer 2 Circuit Service model."""

    class ServiceBindingPortInput(BaseModel):
        """Service Binding Port model."""

        edge_port: UUIDstr
        vlan_id: VLAN_ID

    service_type: Layer2CircuitServiceType
    partner: ExistingPartnerStr
    gs_id: IMPORTED_GS_ID
    vc_id: VC_ID
    layer_2_circuit_side_a: ServiceBindingPortInput
    layer_2_circuit_side_b: ServiceBindingPortInput
    layer_2_circuit_type: Layer2CircuitType
    vlan_range_lower_bound: VLAN_ID | None = None
    vlan_range_upper_bound: VLAN_ID | None = None
    policer_enabled: bool = False
    policer_bandwidth: BandwidthString | None = None
    policer_burst_rate: BandwidthString | None = None
    custom_service_name: str | None = None

    @model_validator(mode="after")
    def check_if_edge_ports_exist(self) -> Self:
        """Check if the edge ports exist."""
        for side in [self.layer_2_circuit_side_a, self.layer_2_circuit_side_b]:
            if not EdgePort.from_subscription(side.edge_port):
                msg = f"Edge Port {side.edge_port} not found"
                raise ValueError(msg)
        return self

ServiceBindingPortInput

Bases: BaseModel

Service Binding Port model.

Source code in gso/cli/imports.py

class ServiceBindingPortInput(BaseModel):
    """Service Binding Port model."""

    edge_port: UUIDstr
    vlan_id: VLAN_ID

check_if_edge_ports_exist()

Check if the edge ports exist.

Source code in gso/cli/imports.py

@model_validator(mode="after")
def check_if_edge_ports_exist(self) -> Self:
    """Check if the edge ports exist."""
    for side in [self.layer_2_circuit_side_a, self.layer_2_circuit_side_b]:
        if not EdgePort.from_subscription(side.edge_port):
            msg = f"Edge Port {side.edge_port} not found"
            raise ValueError(msg)
    return self

PlacementPortImportModel

Bases: BaseModel

Import Placement Port model.

Source code in gso/cli/imports.py

class PlacementPortImportModel(BaseModel):
    """Import Placement Port model."""

    class L3InterfacePort(BaseModel):
        """L3 Interface Port model for Placement Port Import."""

        edge_port: Annotated[str, AfterValidator(validate_edge_port_exists)]
        gs_id: IMPORTED_GS_ID
        is_tagged: bool = False
        vlan_id: VLAN_ID
        custom_firewall_filters: bool = False
        ipv4_address: IPv4AddressType
        ipv4_mask: IPv4Netmask
        ipv6_address: IPv6AddressType
        ipv6_mask: IPv6Netmask

    partner: ExistingPartnerStr
    l3_interface: L3InterfacePort
    product_name: PlacementPortNameType

L3InterfacePort

Bases: BaseModel

L3 Interface Port model for Placement Port Import.

Source code in gso/cli/imports.py

class L3InterfacePort(BaseModel):
    """L3 Interface Port model for Placement Port Import."""

    edge_port: Annotated[str, AfterValidator(validate_edge_port_exists)]
    gs_id: IMPORTED_GS_ID
    is_tagged: bool = False
    vlan_id: VLAN_ID
    custom_firewall_filters: bool = False
    ipv4_address: IPv4AddressType
    ipv4_mask: IPv4Netmask
    ipv6_address: IPv6AddressType
    ipv6_mask: IPv6Netmask

PeeringConnectionImportModel

Bases: BaseModel

Peering connection import model.

Source code in gso/cli/imports.py

class PeeringConnectionImportModel(BaseModel):
    """Peering connection import model."""

    bgp_session_v4: BGPPeerImportModelSimple
    bgp_session_v6: BGPPeerImportModelSimple
    minimum_hold_timer: int | None = None
    session_state: SessionState
    placement_port: UUIDstr

    @model_validator(mode="after")
    def validate_placement_port_exists(self) -> Self:
        """Validate that the placement port exists."""
        subscription = SubscriptionModel.from_subscription(self.placement_port)

        if not subscription:
            msg = f"Placement Port {self.placement_port} not found in the database."
            raise ValueError(msg)

        if subscription.status != SubscriptionLifecycle.ACTIVE:
            msg = f"Placement Port {self.placement_port} is not active."
            raise ValueError(msg)

        return self

validate_placement_port_exists()

Validate that the placement port exists.

Source code in gso/cli/imports.py

@model_validator(mode="after")
def validate_placement_port_exists(self) -> Self:
    """Validate that the placement port exists."""
    subscription = SubscriptionModel.from_subscription(self.placement_port)

    if not subscription:
        msg = f"Placement Port {self.placement_port} not found in the database."
        raise ValueError(msg)

    if subscription.status != SubscriptionLifecycle.ACTIVE:
        msg = f"Placement Port {self.placement_port} is not active."
        raise ValueError(msg)

    return self

CommercialPeerImportModel

Bases: BaseModel

Import Commercial Peer model.

Source code in gso/cli/imports.py

class CommercialPeerImportModel(BaseModel):
    """Import Commercial Peer model."""

    class Config:
        """Pydantic model configuration."""

        extra = "forbid"

    partner: ExistingPartnerStr
    peering_connection_list: list[PeeringConnectionImportModel]
    prefix_limit_v4: NonNegativeInt | None = None
    prefix_limit_v6: NonNegativeInt | None = None
    partner_asn: NonNegativeInt | None = None
    partner_org_name: str | None = None

Config

Pydantic model configuration.

Source code in gso/cli/imports.py

class Config:
    """Pydantic model configuration."""

    extra = "forbid"

PeeringConnectionListImportModel

Bases: BaseModel

Import model for adding peering connections to an existing Commercial Peer subscription.

Source code in gso/cli/imports.py

class PeeringConnectionListImportModel(BaseModel):
    """Import model for adding peering connections to an existing Commercial Peer subscription."""

    peering_connection_list: list[PeeringConnectionImportModel]

PrefixListImportModel

Bases: BaseModel

Import model for adding existing Prefix Lists to the subscription database.

Source code in gso/cli/imports.py

class PrefixListImportModel(BaseModel):
    """Import model for adding existing Prefix Lists to the subscription database."""

    partner_names: list[str]

    @field_validator("partner_names")
    def partners_can_only_have_one_subscription(cls, partner_names: list[str]) -> list[str]:
        """Ensure that one partner does not have multiple Prefix List subscriptions before importing more."""
        for partner_name in partner_names:
            if get_subscriptions(
                product_types=[ProductType.PREFIX_LIST],
                lifecycles=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
                partner_id=get_partner_by_name(partner_name).partner_id,
            ):
                msg = f"Partner {partner_name} already has a Prefix List subscription!"
                raise ValueError(msg)

        return partner_names

partners_can_only_have_one_subscription(partner_names)

Ensure that one partner does not have multiple Prefix List subscriptions before importing more.

Source code in gso/cli/imports.py

@field_validator("partner_names")
def partners_can_only_have_one_subscription(cls, partner_names: list[str]) -> list[str]:
    """Ensure that one partner does not have multiple Prefix List subscriptions before importing more."""
    for partner_name in partner_names:
        if get_subscriptions(
            product_types=[ProductType.PREFIX_LIST],
            lifecycles=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
            partner_id=get_partner_by_name(partner_name).partner_id,
        ):
            msg = f"Partner {partner_name} already has a Prefix List subscription!"
            raise ValueError(msg)

    return partner_names

_read_data(file_path)

Read data from a JSON or YAML file.

Source code in gso/cli/imports.py

def _read_data(file_path: Path) -> dict:
    """Read data from a JSON or YAML file."""
    typer.echo(f"Starting import from {file_path!s}")
    file_extension = file_path.suffix.lower()

    with file_path.open("r") as f:
        supported_extensions = {".json", ".yaml", ".yml"}

        if file_extension == ".json":
            return json.load(f)
        if file_extension in supported_extensions:
            return yaml.safe_load(f)

        typer.echo(f"Unsupported file format: {file_extension}")
        raise typer.Exit(code=1)

_import_partners_from_csv(file_path)

Read partners from a CSV file.

Source code in gso/cli/imports.py

def _import_partners_from_csv(file_path: Path) -> list[dict]:
    """Read partners from a CSV file."""
    with Path.open(file_path, encoding="utf-8") as csv_file:
        csv_reader = csv.DictReader(csv_file)
        return list(csv_reader)

_generic_import_product(file_path, imported_product_type, workflow_suffix, name_key, import_model)

Import subscriptions from a JSON or YAML file.

Source code in gso/cli/imports.py

def _generic_import_product[
    T: (
        SiteImportModel,
        RouterImportModel,
        IptrunkImportModel,
        SuperPopSwitchImportModel,
        OfficeRouterImportModel,
        OpenGearImportModel,
        EdgePortImportModel,
        L3CoreServiceImportModel,
        Layer2CircuitServiceImportModel,
    )
](
    file_path: Path, imported_product_type: ProductType, workflow_suffix: str, name_key: str, import_model: type[T]
) -> None:
    """Import subscriptions from a JSON or YAML file."""
    successfully_imported_data = []
    data = _read_data(file_path)
    for details in data:
        details["partner"] = details.get("partner", "GEANT")
        typer.echo(f"Creating imported {name_key}: {details[name_key]}")
        try:
            initial_data = import_model(**details)
            start_process(f"create_imported_{workflow_suffix}", [initial_data.model_dump()])
            successfully_imported_data.append(getattr(initial_data, name_key))
            typer.echo(
                f"Successfully created {name_key}: {getattr(initial_data, name_key)}",
            )
        except ValidationError as e:
            typer.echo(f"Validation error: {e}")

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    #  Migrate new products from imported to "full" counterpart.
    imported_products = get_subscriptions(
        product_types=[imported_product_type], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=["subscription_id"]
    )
    for subscription_id in imported_products:
        typer.echo(f"Importing {subscription_id}")
        start_process(f"import_{workflow_suffix}", [subscription_id])

    if successfully_imported_data:
        typer.echo(f"Successfully created imported {name_key}s:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")
        typer.echo(f"Please validate no more imported {workflow_suffix} products exist anymore in the database.")

import_sites(filepath=common_filepath_option)

Import sites into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_sites(filepath: str = common_filepath_option) -> None:
    """Import sites into GSO."""
    _generic_import_product(Path(filepath), ProductType.IMPORTED_SITE, "site", "site_name", SiteImportModel)

import_routers(filepath=common_filepath_option)

Import routers into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_routers(filepath: str = common_filepath_option) -> None:
    """Import routers into GSO."""
    _generic_import_product(Path(filepath), ProductType.IMPORTED_ROUTER, "router", "hostname", RouterImportModel)

import_super_pop_switches(filepath=common_filepath_option)

Import Super PoP Switches into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_super_pop_switches(filepath: str = common_filepath_option) -> None:
    """Import Super PoP Switches into GSO."""
    _generic_import_product(
        Path(filepath),
        ProductType.IMPORTED_SUPER_POP_SWITCH,
        "super_pop_switch",
        "hostname",
        SuperPopSwitchImportModel,
    )

import_office_routers(filepath=common_filepath_option)

Import office routers into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_office_routers(filepath: str = common_filepath_option) -> None:
    """Import office routers into GSO."""
    _generic_import_product(
        Path(filepath),
        ProductType.IMPORTED_OFFICE_ROUTER,
        "office_router",
        "office_router_fqdn",
        OfficeRouterImportModel,
    )

import_opengear(filepath=common_filepath_option)

Import Opengear into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_opengear(filepath: str = common_filepath_option) -> None:
    """Import Opengear into GSO."""
    _generic_import_product(
        Path(filepath),
        ProductType.IMPORTED_OPENGEAR,
        "opengear",
        "opengear_hostname",
        OpenGearImportModel,
    )

import_edge_port(filepath=common_filepath_option)

Import Edge Port into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_edge_port(filepath: str = common_filepath_option) -> None:
    """Import Edge Port into GSO."""
    successfully_imported_data = []
    data = _read_data(Path(filepath))
    for edge_port in data:
        typer.echo(f"Importing Edge Port {edge_port['name']} on {edge_port['node']}. ")
        try:
            edge_port["node"] = get_active_router_subscription_id(edge_port["node"])
            initial_data = EdgePortImportModel(**edge_port)
            start_process("create_imported_edge_port", [initial_data.model_dump()])
            successfully_imported_data.append(edge_port["name"])
            typer.echo(f"Successfully imported Edge Port {edge_port['name']} on {edge_port['node']}.")
        except ValidationError as e:
            typer.echo(f"Validation error: {e}")

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    edge_port_ids = get_subscriptions(
        product_types=[ProductType.IMPORTED_EDGE_PORT],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id"],
    )
    for subscription_id in edge_port_ids:
        typer.echo(f"Migrating Edge Port {subscription_id}")
        start_process("import_edge_port", [subscription_id])

    if successfully_imported_data:
        typer.echo("Successfully imported Edge Ports:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")

import_iptrunks(filepath=common_filepath_option)

Import IP trunks into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_iptrunks(filepath: str = common_filepath_option) -> None:
    """Import IP trunks into GSO."""

    def _read_and_validate_networks(trunk: dict) -> tuple[ipaddress.IPv4Network, ipaddress.IPv6Network] | None:
        """Validate that nodeA and nodeB share the same IPv4 and IPv6 networks.

        Returns (ipv4_network, ipv6_network) or None if validation fails.
        """
        ipv4_a = ipaddress.IPv4Network(trunk["config"]["nodeA"]["ipv4_address"], strict=False)
        ipv4_b = ipaddress.IPv4Network(trunk["config"]["nodeB"]["ipv4_address"], strict=False)
        if ipv4_a != ipv4_b:
            typer.echo(f"Error: IPv4 networks differ for trunk {trunk['id']}.")
            return None

        ipv6_a = ipaddress.IPv6Network(trunk["config"]["nodeA"]["ipv6_address"], strict=False)
        ipv6_b = ipaddress.IPv6Network(trunk["config"]["nodeB"]["ipv6_address"], strict=False)
        if ipv6_a != ipv6_b:
            typer.echo(f"Error: IPv6 networks differ for trunk {trunk['id']}.")
            return None

        return ipv4_a, ipv6_a

    def _build_import_model(
        trunk: dict, ipv4_net: ipaddress.IPv4Network, ipv6_net: ipaddress.IPv6Network
    ) -> IptrunkImportModel:
        """Constructs the Pydantic model for an IP trunk import."""
        return IptrunkImportModel(
            partner="GEANT",
            gs_id=trunk["id"],
            iptrunk_type=trunk["config"]["common"]["type"],
            iptrunk_description=trunk["config"]["common"].get("description", ""),
            iptrunk_speed=trunk["config"]["common"]["link_speed"],
            iptrunk_minimum_links=trunk["config"]["common"]["minimum_links"],
            iptrunk_isis_metric=trunk["config"]["common"]["isis_metric"],
            side_a_node_id=get_active_router_subscription_id(trunk["config"]["nodeA"]["name"]) or "",
            side_a_ae_iface=trunk["config"]["nodeA"]["ae_name"],
            side_a_ga_id=trunk["config"]["nodeA"]["port_ga_id"],
            side_a_ae_members=trunk["config"]["nodeA"]["members"],
            side_b_node_id=get_active_router_subscription_id(trunk["config"]["nodeB"]["name"]) or "",
            side_b_ae_iface=trunk["config"]["nodeB"]["ae_name"],
            side_b_ga_id=trunk["config"]["nodeB"]["port_ga_id"],
            side_b_ae_members=trunk["config"]["nodeB"]["members"],
            iptrunk_ipv4_network=ipv4_net,
            iptrunk_ipv6_network=ipv6_net,
        )

    def _import_single_trunk(trunk: dict) -> bool:
        """Attempt import of one trunk. Returns True on success."""
        validated = _read_and_validate_networks(trunk)
        if not validated:
            return False
        ipv4_net, ipv6_net = validated

        typer.echo(
            f"Importing IP Trunk: "
            f"{get_active_subscriptions_by_field_and_value('router_fqdn', trunk['config']['nodeA']['name'])}"
        )

        try:
            model = _build_import_model(trunk, ipv4_net, ipv6_net)
            start_process("create_imported_iptrunk", [model.model_dump()])
        except ValidationError as e:
            typer.echo(f"Validation error: {e}")
            return False

        typer.echo(f"Successfully imported IP Trunk: {trunk['id']}")
        return True

    def _create_imported() -> list[str]:
        data = _read_data(Path(filepath))

        # First pass: create all the imported trunks
        imported_ids = [trunk["id"] for trunk in data if _import_single_trunk(trunk)]

        typer.echo(IMPORT_WAIT_MESSAGE)
        time.sleep(_SLEEP_TIME_SECONDS)

        # Second pass: kick off the migration processes
        for sub in get_subscriptions(
            product_types=[ProductType.IMPORTED_IP_TRUNK],
            lifecycles=[SubscriptionLifecycle.ACTIVE],
            includes=["subscription_id"],
        ):
            typer.echo(f"Migrating iptrunk {sub}")
            start_process("import_iptrunk", [sub])

        return imported_ids

    successfully_imported_data = _create_imported()

    if successfully_imported_data:
        typer.echo("Successfully imported IP Trunks:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")

import_partners(file_path=typer.Argument(..., help='Path to the CSV file containing partners'))

Import partners from a CSV file into the database.

Source code in gso/cli/imports.py

@app.command()
def import_partners(file_path: str = typer.Argument(..., help="Path to the CSV file containing partners")) -> None:
    """Import partners from a CSV file into the database."""
    typer.echo(f"Importing partners from {file_path} ...")

    partners = _import_partners_from_csv(Path(file_path))

    try:
        for partner in partners:
            if partner.get("created_at"):
                partner["created_at"] = datetime.strptime(partner["created_at"], "%Y-%m-%d").replace(tzinfo=UTC)

            new_partner = PartnerTable(**CreatePartner(**partner).model_dump())
            db.session.add(new_partner)

        db.session.commit()
        typer.echo(f"Successfully imported {len(partners)} partners.")
    except SQLAlchemyError as e:
        db.session.rollback()
        typer.echo(f"Failed to import partners: {e}")
    finally:
        db.session.close()

import_l3_core_service(filepath=common_filepath_option)

Import L3 Core Services into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_l3_core_service(filepath: str = common_filepath_option) -> None:
    """Import L3 Core Services into GSO."""

    def _create_imported() -> list[list[str]]:
        imported_ids = []
        l3_core_service_list = _read_data(Path(filepath))
        for l3_core_service in l3_core_service_list:
            partner = l3_core_service["partner"]
            p_name = l3_core_service["product_name"]
            typer.echo(f"Creating imported {p_name} for {partner}")

            try:
                if p_name == ProductName.IAS.value:
                    initial_data = IASImportModel(**l3_core_service).model_dump()
                elif p_name in {ProductName.R_AND_E_PEER.value, ProductName.R_AND_E_LHCONE.value}:
                    initial_data = RAndEImportModel(**l3_core_service).model_dump()
                else:
                    initial_data = L3CoreServiceImportModel(**l3_core_service).model_dump()

                start_process(L3_CREATE_IMPORTED_WF_MAP[p_name], [initial_data])
                edge_ports = [sbp["edge_port"] for sbp in l3_core_service["service_binding_ports"]]
                imported_ids.append(edge_ports)
                typer.echo(f"Successfully created imported {p_name} for {partner}")
            except ValidationError as e:
                typer.echo(f"Validation error: {e}")

        return imported_ids

    successfully_imported_data = _create_imported()

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    # Migrate new products from imported to "full" counterpart.
    imported_products = get_subscriptions(
        product_types=[
            ProductType.IMPORTED_GEANT_IP,
            ProductType.IMPORTED_IAS,
            ProductType.IMPORTED_LHCONE,
            ProductType.IMPORTED_COPERNICUS,
            ProductType.IMPORTED_R_AND_E_PEER,
            ProductType.IMPORTED_R_AND_E_LHCONE,
        ],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id", "product_id"],
    )

    for subscription in imported_products:
        product_name = get_product_by_id(subscription["product_id"]).name  # type: ignore[union-attr]
        typer.echo(f"Importing {product_name} with {subscription['product_id']}")
        start_process(L3_IMPORT_WF_MAP[product_name], [{"subscription_id": subscription["subscription_id"]}])

    if successfully_imported_data:
        typer.echo("Successfully created imported L3 Core Services:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")

import_layer_2_circuit_service(filepath=common_filepath_option)

Import Layer 2 Circuit services into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_layer_2_circuit_service(filepath: str = common_filepath_option) -> None:
    """Import Layer 2 Circuit services into GSO."""
    successfully_imported_data = []
    layer_2_circuit_service_list = _read_data(Path(filepath))

    for layer_2_circuit_service in layer_2_circuit_service_list:
        partner = layer_2_circuit_service["partner"]
        service_type = Layer2CircuitServiceType(layer_2_circuit_service["service_type"])
        typer.echo(f"Creating imported {service_type} for {partner}")

        try:
            initial_data = Layer2CircuitServiceImportModel(**layer_2_circuit_service)
            start_process("create_imported_layer_2_circuit", [initial_data.model_dump()])
            successfully_imported_data.append(initial_data.vc_id)
            typer.echo(
                f"Successfully created imported {service_type} with virtual circuit ID {initial_data.vc_id}"
                f" for {partner}"
            )
        except ValidationError as e:
            typer.echo(f"Validation error: {e}")

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    # Migrate new products from imported to "full" counterpart.
    imported_products = get_subscriptions(
        product_types=[ProductType.IMPORTED_EXPRESSROUTE, ProductType.IMPORTED_GEANT_PLUS],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id"],
    )

    for subscription_id in imported_products:
        typer.echo(f"Importing {subscription_id}")
        start_process("import_layer_2_circuit", [subscription_id])

    if successfully_imported_data:
        typer.echo("Successfully created imported Layer 2 Circuit services:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")

import_placement_port(filepath=common_filepath_option)

Import Placement Port(IX Port, Private Peer Port, and Transit Provider Port) into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_placement_port(filepath: str = common_filepath_option) -> None:
    """Import Placement Port(IX Port, Private Peer Port, and Transit Provider Port) into GSO."""
    successfully_imported_data = []
    placement_port_list = _read_data(Path(filepath))

    for placement_port in placement_port_list:
        partner = placement_port["partner"]
        product_name = placement_port["product_name"]
        typer.echo(f"Creating imported {product_name} for {partner}")

        try:
            initial_data = PlacementPortImportModel(**placement_port).model_dump()

            start_process(PLACEMENT_PORT_CREATE_IMPORTED_WF_MAP[product_name], [initial_data])
            successfully_imported_data.append(placement_port["l3_interface"]["edge_port"])
            typer.echo(f"Successfully created imported {product_name} for {partner}")
        except ValidationError as e:
            typer.echo(f"Validation error: {e}")

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    # Migrate new products from imported to "full" counterpart.
    imported_products = get_subscriptions(
        product_types=[
            ProductType.IMPORTED_IX_PORT,
            ProductType.IMPORTED_PRIVATE_PEER_PORT,
            ProductType.IMPORTED_R_AND_E_PEER,
        ],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id", "product_id"],
    )

    for subscription in imported_products:
        product_name = get_product_by_id(subscription["product_id"]).name  # type: ignore[union-attr]
        typer.echo(f"Importing {product_name} with {subscription['product_id']}")
        start_process(
            PLACEMENT_PORT_IMPORT_WF_MAP[product_name], [{"subscription_id": subscription["subscription_id"]}]
        )

    if successfully_imported_data:
        typer.echo("Successfully created imported L3 Core Services:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")

import_commercial_peer(filepath=common_filepath_option)

Import Commercial Peer into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_commercial_peer(filepath: str = common_filepath_option) -> None:
    """Import Commercial Peer into GSO."""
    successfully_imported_data = []
    commercial_peers = _read_data(Path(filepath))
    for commercial_peer in commercial_peers:
        partner_name = commercial_peer.get("partner", "<missing>")
        try:
            partner = get_partner_by_name(partner_name)
        except PartnerNotFoundError:
            typer.echo(f"Validation error: Partner {partner_name} not found.")
            continue

        if commercial_peer_subscription := get_active_commercial_peer_by_partner_id(partner.partner_id):
            typer.echo(f"Commercial Peer for {partner_name} already exists, importing peering connections only.")

            input_data = [
                {"subscription_id": commercial_peer_subscription.subscription_id},
                PeeringConnectionListImportModel(
                    peering_connection_list=commercial_peer["peering_connection_list"]
                ).model_dump(),
            ]
            start_process("import_peering_connections", input_data)
        else:
            typer.echo(f"Creating imported Commercial Peer for {partner_name}")
            try:
                start_process(
                    "create_imported_commercial_peer", [CommercialPeerImportModel(**commercial_peer).model_dump()]
                )
                successfully_imported_data.append(f"Commercial Peer - {partner_name}")
                typer.echo(
                    f"Successfully created Commercial Peer for {partner_name}",
                )
            except ValidationError as e:
                typer.echo(f"Validation error: {e}")

    typer.echo(IMPORT_WAIT_MESSAGE)
    time.sleep(_SLEEP_TIME_SECONDS)

    imported_products = get_subscriptions(
        product_types=[ProductType.IMPORTED_COMMERCIAL_PEER],
        lifecycles=[SubscriptionLifecycle.ACTIVE],
        includes=["subscription_id"],
    )
    for subscription_id in imported_products:
        typer.echo(f"Importing {subscription_id}")
        start_process("import_commercial_peer", [subscription_id])

    if successfully_imported_data:
        typer.echo("Successfully created imported commercial peers:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")
        typer.echo("Please validate no more imported commercial peer products exist anymore in the database.")

import_prefix_list(filepath=common_filepath_option)

Import one or more Prefix Lists into GSO.

Source code in gso/cli/imports.py

@app.command()
def import_prefix_list(filepath: str = common_filepath_option) -> None:
    """Import one or more Prefix Lists into GSO."""
    successfully_imported_data = []
    failed_imported_data = []
    import_data = PrefixListImportModel(**_read_data(Path(filepath)))
    for partner in import_data.partner_names:
        typer.echo(f"Creating Prefix List for {partner}")
        user_inputs = [
            {"product": get_product_id_by_name(ProductName.PREFIX_LIST)},
            {"partner": get_partner_by_name(partner).partner_id},
        ]
        try:
            start_process("create_prefix_list", user_inputs)
            successfully_imported_data.append(f"Prefix List for {partner}")
            typer.echo(f"Successfully created Prefix List for {partner}")
        except FormValidationError as e:
            failed_imported_data.append(f"Prefix List for {partner}")
            typer.echo(f"Form validation error: {e}")

    if failed_imported_data:
        typer.echo("Failed to create some Prefix Lists:")
        for item in failed_imported_data:
            typer.echo(f"- {item}")
    if successfully_imported_data:
        typer.echo("Successfully created Prefix Lists:")
        for item in successfully_imported_data:
            typer.echo(f"- {item}")