diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py index c20e8a4e780daff69688fa1b83b7cf57787ce99a..8b0bfa248934081998ef702d122b09cb6a198d72 100644 --- a/gso/utils/helpers.py +++ b/gso/utils/helpers.py @@ -34,7 +34,7 @@ if TYPE_CHECKING: from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock -def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None: +def available_interfaces_choices(router_id: UUID, speed: str) -> TypeAlias: """Return a list of available interfaces for a given router and speed. For Nokia routers, return a list of available interfaces. @@ -46,14 +46,17 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None: interface["name"]: f"{interface["name"]} {interface["description"]}" for interface in NetboxClient().get_available_interfaces(router_id, speed) } - return Choice("ae member", zip(interfaces.keys(), interfaces.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("ae member", zip(interfaces.keys(), interfaces.items(), strict=True)), + ) def available_interfaces_choices_including_current_members( router_id: UUID, speed: str, interfaces: list["IptrunkInterfaceBlock"], -) -> Choice | None: +) -> TypeAlias: """Return a list of available interfaces for a given router and speed including the current members. For Nokia routers, return a list of available interfaces. @@ -75,10 +78,13 @@ def available_interfaces_choices_including_current_members( options = { interface["name"]: f"{interface["name"]} {interface["description"]}" for interface in available_interfaces } - return Choice("ae member", zip(options.keys(), options.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("ae member", zip(options.keys(), options.items(), strict=True)), + ) -def available_lags_choices(router_id: UUID) -> Choice | None: +def available_lags_choices(router_id: UUID) -> TypeAlias: """Return a list of available lags for a given router. For Nokia routers, return a list of available lags. @@ -87,10 +93,13 @@ def available_lags_choices(router_id: UUID) -> Choice | None: if get_router_vendor(router_id) != Vendor.NOKIA: return None side_a_ae_iface_list = NetboxClient().get_available_lags(router_id) - return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)), + ) -def available_service_lags_choices(router_id: UUID) -> Choice | None: +def available_service_lags_choices(router_id: UUID) -> TypeAlias: """Return a list of available lags for a given router for services. For Nokia routers, return a list of available lags. @@ -99,7 +108,10 @@ def available_service_lags_choices(router_id: UUID) -> Choice | None: if get_router_vendor(router_id) != Vendor.NOKIA: return None side_a_ae_iface_list = NetboxClient().get_available_services_lags(router_id) - return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list, strict=True)), + ) def get_router_vendor(router_id: UUID) -> Vendor: @@ -222,30 +234,60 @@ def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_ return iptrunk_number_of_members -def active_site_selector() -> Choice: +def active_site_selector() -> TypeAlias: """Generate a dropdown selector for choosing an active site in an input form.""" site_subscriptions = { str(site["subscription_id"]): site["description"] for site in get_active_site_subscriptions(includes=["subscription_id", "description"]) } - return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True)), + ) + +def active_router_selector(*, excludes: list[UUIDstr] | None = None) -> TypeAlias: + """Generate a dropdown selector for choosing an active Router in an input form. -def active_router_selector(*, excludes: list[UUIDstr] | None = None) -> Choice: - """Generate a dropdown selector for choosing an active Router in an input form.""" - if excludes is None: - excludes = [] + The resulting list of routers can be filtered using a list of excluded subscription IDs. + """ + excludes = excludes or [] router_subscriptions = { str(router["subscription_id"]): router["description"] for router in get_active_router_subscriptions(includes=["subscription_id", "description"]) if router["subscription_id"] not in excludes } - return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)), + ) + +def active_nokia_router_selector(*, excludes: list[UUIDstr] | None = None) -> TypeAlias: + """Generate a dropdown choice list of all active Nokia routers. -def active_pe_router_selector(excludes: list[UUIDstr] | None = None) -> Choice: + Args: + excludes: An optional list of subscription IDs that should be excluded from the resulting dropdown. + """ + excludes = excludes or [] + router_subscriptions = { + str(router.subscription_id): router.description + for router in [ + Router.from_subscription(subscription["subscription_id"]) + for subscription in get_active_router_subscriptions(["subscription_id"]) + ] + if router.subscription_id not in excludes and router.router.vendor == Vendor.NOKIA + } + + return cast( + type[Choice], + Choice.__call__("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True)), + ) + + +def active_pe_router_selector(excludes: list[UUIDstr] | None = None) -> TypeAlias: """Generate a dropdown selector for choosing an active PE Router in an input form.""" excludes = excludes or [] @@ -255,17 +297,23 @@ def active_pe_router_selector(excludes: list[UUIDstr] | None = None) -> Choice: if router.subscription_id not in excludes } - return Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("Select a router", zip(routers.keys(), routers.items(), strict=True)), + ) -def active_switch_selector() -> Choice: +def active_switch_selector() -> TypeAlias: """Generate a dropdown selector for choosing an active Switch in an input form.""" switch_subscriptions = { str(switch["subscription_id"]): switch["description"] for switch in get_active_switch_subscriptions(includes=["subscription_id", "description"]) } - return Choice("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("Select a switch", zip(switch_subscriptions.keys(), switch_subscriptions.items(), strict=True)), + ) def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> TypeAlias: @@ -283,11 +331,14 @@ def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> TypeAlias ) -def partner_choice() -> Choice: +def partner_choice() -> TypeAlias: """Return a Choice object containing a list of available partners.""" partners = {partner.partner_id: partner.name for partner in get_all_partners()} - return Choice("Select a partner", zip(partners.values(), partners.items(), strict=True)) # type: ignore[arg-type] + return cast( + type[Choice], + Choice.__call__("Select a partner", zip(partners.values(), partners.items(), strict=True)), + ) def validate_edge_port_number_of_members_based_on_lacp(*, number_of_members: int, enable_lacp: bool) -> None: @@ -315,7 +366,7 @@ def generate_unique_vc_id(l2c_type: str, max_attempts: int = 100) -> VC_ID | Non ``Ethernet`` and ``VLAN`` type circuits get their IDs from different ranges. Args: - l2c_type: type of l2circuit. + l2c_type: type of Layer 2 Circuit. max_attempts: The maximum number of attempts to generate a unique ID. Returns: