Newer
Older
import ipaddress
from typing import Any
from uuid import UUID
from pydantic import BaseModel, root_validator, validator
from gso.products.product_blocks import PhyPortCapacity
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole, RouterVendor
from gso.products.product_blocks.site import SiteTier
from gso.services import subscriptions
from gso.services.crm import CustomerNotFoundError, get_customer_by_name
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
class ImportResponseModel(BaseModel):
pid: UUID
detail: str
class SiteImportModel(BaseModel):
site_name: str
site_city: str
site_country: str
site_country_code: str
site_latitude: float
site_longitude: float
site_bgp_community_id: int
site_internal_id: int
site_tier: SiteTier
site_ts_address: str
customer: str
class RouterImportModel(BaseModel):
customer: str
router_site: str
hostname: str
ts_port: int
router_vendor: RouterVendor
router_role: RouterRole
is_ias_connected: bool
router_lo_ipv4_address: ipaddress.IPv4Address
router_lo_ipv6_address: ipaddress.IPv6Address
router_lo_iso_address: str
router_si_ipv4_network: ipaddress.IPv4Network | None = None
router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
class IptrunkImportModel(BaseModel):
customer: str
geant_s_sid: str
iptrunk_type: IptrunkType
iptrunk_description: str
iptrunk_speed: PhyPortCapacity
iptrunk_minimum_links: int
iptrunk_sideA_node_id: str
iptrunk_sideA_ae_iface: str
iptrunk_sideA_ae_geant_a_sid: str
iptrunk_sideA_ae_members: list[str]
iptrunk_sideA_ae_members_descriptions: list[str]
iptrunk_sideB_node_id: str
iptrunk_sideB_ae_iface: str
iptrunk_sideB_ae_geant_a_sid: str
iptrunk_sideB_ae_members: list[str]
iptrunk_sideB_ae_members_descriptions: list[str]
iptrunk_ipv4_network: ipaddress.IPv4Network
iptrunk_ipv6_network: ipaddress.IPv6Network
@classmethod
def _get_active_routers(cls) -> set[str]:
return {
str(router_id) for router_id in subscriptions.get_active_router_subscriptions(fields=["subscription_id"])
}
@validator("customer")
def check_if_customer_exists(cls, value: str) -> str:
try:
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
except CustomerNotFoundError:
raise ValueError(f"Customer {value} not found")
return value
@validator("iptrunk_sideA_node_id", "iptrunk_sideB_node_id")
def check_if_router_side_is_available(cls, value: str) -> str:
if value not in cls._get_active_routers():
raise ValueError("Router not found")
return value
@validator("iptrunk_sideA_ae_members", "iptrunk_sideB_ae_members")
def check_side_uniqueness(cls, value: list[str]) -> list[str]:
if len(value) != len(set(value)):
raise ValueError("Items must be unique")
return value
@root_validator
def check_members(cls, values: dict[str, Any]) -> dict[str, Any]:
min_links = values["iptrunk_minimum_links"]
side_a_members = values.get("iptrunk_sideA_ae_members", [])
side_a_descriptions = values.get("iptrunk_sideA_ae_members_descriptions", [])
side_b_members = values.get("iptrunk_sideB_ae_members", [])
side_b_descriptions = values.get("iptrunk_sideB_ae_members_descriptions", [])
len_a = len(side_a_members)
len_a_desc = len(side_a_descriptions)
len_b = len(side_b_members)
len_b_desc = len(side_b_descriptions)
if len_a < min_links:
raise ValueError(f"Side A members should be at least {min_links} (iptrunk_minimum_links)")
if len_a != len_a_desc:
raise ValueError("Mismatch in Side A members and their descriptions")
if len_a != len_b:
raise ValueError("Mismatch between Side A and B members")
if len_a != len_b_desc:
raise ValueError("Mismatch in Side B members and their descriptions")
return values