network.py

"""API endpoints for network related operations."""

from uuid import UUID

from fastapi import APIRouter
from orchestrator.domain import SubscriptionModel
from orchestrator.schemas.base import OrchestratorBaseModel
from orchestrator.services.subscriptions import build_extended_domain_model
from starlette import status

from gso.products.product_blocks.iptrunk import PhysicalPortCapacity
from gso.services.subscriptions import get_active_iptrunk_subscriptions

router = APIRouter(prefix="/networks", tags=["Network"])


class RouterBlock(OrchestratorBaseModel):
    """Router block schema."""

    subscription_instance_id: UUID
    router_fqdn: str


class IptrunkSideBlock(OrchestratorBaseModel):
    """Iptrunk side block schema."""

    subscription_instance_id: UUID
    iptrunk_side_node: RouterBlock


class IptrunkBlock(OrchestratorBaseModel):
    """Iptrunk block schema."""

    subscription_instance_id: UUID
    iptrunk_speed: str
    iptrunk_capacity: str
    iptrunk_isis_metric: int
    iptrunk_sides: list[IptrunkSideBlock]


class IptrunkSchema(OrchestratorBaseModel):
    """Iptrunk schema."""

    subscription_id: UUID
    insync: bool
    iptrunk: IptrunkBlock


class NetworkTopologyDomainModelSchema(OrchestratorBaseModel):
    """Network topology domain model schema."""

    iptrunks: list[IptrunkSchema]


def _calculate_iptrunk_capacity(iptrunk_sides: list, iptrunk_speed: PhysicalPortCapacity) -> str:
    """Calculate the total capacity of an IP trunk."""
    int_iptrunk_speed = int(iptrunk_speed.value.replace("G", ""))
    capacity = int_iptrunk_speed * len(iptrunk_sides[0]["iptrunk_side_ae_members"])
    return f"{capacity}G"


@router.get("/topology", status_code=status.HTTP_200_OK, response_model=NetworkTopologyDomainModelSchema)
def network_topology() -> NetworkTopologyDomainModelSchema:
    """Retrieve all active or provisioning IP trunk subscriptions."""
    topology: dict = {"iptrunks": []}
    active_iptrunks = get_active_iptrunk_subscriptions()
    for iptrunk in active_iptrunks:
        subscription = SubscriptionModel.from_subscription(iptrunk["subscription_id"])
        extended_model = build_extended_domain_model(subscription)
        formatted_model = {
            "subscription_id": extended_model["subscription_id"],
            "insync": extended_model["insync"],
            "iptrunk": {
                "subscription_instance_id": extended_model["iptrunk"]["subscription_instance_id"],
                "iptrunk_speed": extended_model["iptrunk"]["iptrunk_speed"],
                "iptrunk_isis_metric": extended_model["iptrunk"]["iptrunk_isis_metric"],
                "iptrunk_capacity": _calculate_iptrunk_capacity(
                    extended_model["iptrunk"]["iptrunk_sides"], extended_model["iptrunk"]["iptrunk_speed"]
                ),
                "iptrunk_sides": [
                    {
                        "subscription_instance_id": side["subscription_instance_id"],
                        "iptrunk_side_node": {
                            "subscription_instance_id": side["iptrunk_side_node"]["subscription_instance_id"],
                            "router_fqdn": side["iptrunk_side_node"]["router_fqdn"],
                        },
                    }
                    for side in extended_model["iptrunk"]["iptrunk_sides"]
                ],
            },
        }
        topology["iptrunks"].append(IptrunkSchema(**formatted_model))

    return NetworkTopologyDomainModelSchema(**topology)