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 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: get_customer_by_name(value) 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