Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
"""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 iptrunk 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)