Finished release 2.13.

.gitignore

@@ -20,3 +20,4 @@ docs/vale/styles/*
 
 .idea
 .venv
+.env
\ No newline at end of file

Changelog.md

 # Changelog
 
+## [2.13] - 2024-09-18
+- Run cleanup at 1 AM only, not every minute between 1 and 2 AM.
+- Add checklist to trunk migration and include ACTIVITY_TYPE column as discussed with SM.
+- Fix SDP update logic in the promotion from P to PE.
+- Update core lib to 2.7.4
+- Refactor validators into a separate types module.
+- Add Tier3 model for Netbox.
+- Fix: Increase initial connection timeout for LibreNMS client.
+- Fix: Change type from LAGMember to dict for workflow step in trunk modification
+- Make celery concurrency level to 1
+
 ## [2.12] - 2024-08-22
 - Add promote P to PE workflow.
 - Add new cleanup task

gso/api/v1/network.py

@@ -10,11 +10,12 @@ from orchestrator.security import authorize
 from orchestrator.services.subscriptions import build_extended_domain_model
 from starlette import status
 
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate
 from gso.services.subscriptions import get_active_iptrunk_subscriptions
 from gso.utils.shared_enums import Vendor
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.interfaces import PhysicalPortCapacity
 
 router = APIRouter(prefix="/networks", tags=["Network"], dependencies=[Depends(authorize)])
 

gso/cli/imports.py

@@ -13,17 +13,17 @@ import yaml
 from orchestrator.db import db
 from orchestrator.services.processes import start_process
 from orchestrator.types import SubscriptionLifecycle
-from pydantic import BaseModel, EmailStr, ValidationError, field_validator, model_validator
+from pydantic import BaseModel, ValidationError, field_validator, model_validator
 from sqlalchemy.exc import SQLAlchemyError
 
 from gso.db.models import PartnerTable
 from gso.products import ProductType
-from gso.products.product_blocks.iptrunk import IptrunkType, PhysicalPortCapacity
+from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
 from gso.services.partners import (
+    PartnerEmail,
+    PartnerName,
     PartnerNotFoundError,
-    filter_partners_by_email,
-    filter_partners_by_name,
     get_partner_by_name,
 )
 from gso.services.subscriptions import (
@@ -31,8 +31,10 @@ from gso.services.subscriptions import (
     get_active_subscriptions_by_field_and_value,
     get_subscriptions,
 )
-from gso.utils.helpers import BaseSiteValidatorModel, LAGMember
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.base_site import BaseSiteValidatorModel
+from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
 
 app: typer.Typer = typer.Typer()
 
@@ -40,27 +42,8 @@ app: typer.Typer = typer.Typer()
 class CreatePartner(BaseModel):
     """Required inputs for creating a partner."""
 
-    name: str
-    email: EmailStr
-
-    @field_validator("name")
-    def validate_name(cls, name: str) -> str:
-        """Validate name."""
-        if filter_partners_by_name(name=name, case_sensitive=False):
-            msg = "Partner with this name already exists."
-            raise ValueError(msg)
-
-        return name
-
-    @field_validator("email")
-    def validate_email(cls, email: str) -> EmailStr:
-        """Validate email."""
-        email = email.lower()
-        if filter_partners_by_email(email=email, case_sensitive=False):
-            msg = "Partner with this email already exists."
-            raise ValueError(msg)
-
-        return email
+    name: PartnerName
+    email: PartnerEmail
 
 
 class SiteImportModel(BaseSiteValidatorModel):
@@ -115,11 +98,11 @@ class IptrunkImportModel(BaseModel):
     side_a_node_id: str
     side_a_ae_iface: str
     side_a_ae_geant_a_sid: str | None
-    side_a_ae_members: list[LAGMember]
+    side_a_ae_members: LAGMemberList[LAGMember]
     side_b_node_id: str
     side_b_ae_iface: str
     side_b_ae_geant_a_sid: str | None
-    side_b_ae_members: list[LAGMember]
+    side_b_ae_members: LAGMemberList[LAGMember]
 
     iptrunk_ipv4_network: ipaddress.IPv4Network
     iptrunk_ipv6_network: ipaddress.IPv6Network
@@ -150,15 +133,6 @@ class IptrunkImportModel(BaseModel):
 
         return value
 
-    @field_validator("side_a_ae_members", "side_b_ae_members")
-    def check_side_uniqueness(cls, value: list[str]) -> list[str]:
-        """:term:`LAG` members must be unique."""
-        if len(value) != len(set(value)):
-            msg = "Items must be unique"
-            raise ValueError(msg)
-
-        return value
-
     @model_validator(mode="after")
     def check_members(self) -> Self:
         """Amount of :term:`LAG` members has to match on side A and B, and meet the minimum requirement."""

gso/products/product_blocks/iptrunk.py

@@ -15,18 +15,7 @@ from gso.products.product_blocks.router import (
     RouterBlockInactive,
     RouterBlockProvisioning,
 )
-
-
-class PhysicalPortCapacity(strEnum):
-    """Physical port capacity enumerator.
-
-    An enumerator that has the different possible capacities of ports that are available to use in subscriptions.
-    """
-
-    ONE_GIGABIT_PER_SECOND = "1G"
-    TEN_GIGABIT_PER_SECOND = "10G"
-    HUNDRED_GIGABIT_PER_SECOND = "100G"
-    FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
+from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
 
 
 class IptrunkType(strEnum):
@@ -36,9 +25,6 @@ class IptrunkType(strEnum):
     LEASED = "Leased"
 
 
-LAGMemberList = Annotated[
-    list[T], AfterValidator(validate_unique_list), Len(min_length=0), Doc("A list of :term:`LAG` member interfaces.")
-]
 IptrunkSides = Annotated[
     list[T],
     AfterValidator(validate_unique_list),

gso/products/product_blocks/lan_switch_interconnect.py

@@ -3,9 +3,9 @@
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
 
-from gso.products.product_blocks.iptrunk import LAGMemberList
 from gso.products.product_blocks.router import RouterBlock, RouterBlockInactive, RouterBlockProvisioning
 from gso.products.product_blocks.switch import SwitchBlock, SwitchBlockInactive, SwitchBlockProvisioning
+from gso.utils.types.interfaces import LAGMemberList
 
 
 class LanSwitchInterconnectAddressSpace(strEnum):

gso/products/product_blocks/office_router.py

@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
 
 
 class OfficeRouterBlockInactive(

gso/products/product_blocks/router.py

@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType, PortNumber
 
 
 class RouterRole(strEnum):

gso/products/product_blocks/site.py

 """The product block that describes a site subscription."""
 
-import re
-from typing import Annotated
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
-from pydantic import AfterValidator
-from typing_extensions import Doc
-
-MAX_LONGITUDE = 180
-MIN_LONGITUDE = -180
 
-MAX_LATITUDE = 90
-MIN_LATITUDE = -90
+from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
+from gso.utils.types.ip_address import IPAddress
+from gso.utils.types.site_name import SiteName
 
 
 class SiteTier(strEnum):
@@ -28,56 +21,6 @@ class SiteTier(strEnum):
     TIER4 = 4
 
 
-def validate_latitude(v: str) -> str:
-    """Validate a latitude coordinate."""
-    msg = "Invalid latitude coordinate. Valid examples: '40.7128', '-74.0060', '90', '-90', '0'."
-    regex = re.compile(r"^-?([1-8]?\d(\.\d+)?|90(\.0+)?)$")
-    if not regex.match(str(v)):
-        raise ValueError(msg)
-
-    float_v = float(v)
-    if float_v > MAX_LATITUDE or float_v < MIN_LATITUDE:
-        raise ValueError(msg)
-
-    return v
-
-
-def validate_longitude(v: str) -> str:
-    """Validate a longitude coordinate."""
-    regex = re.compile(r"^-?(180(\.0+)?|((1[0-7]\d)|([1-9]?\d))(\.\d+)?)$")
-    msg = "Invalid longitude coordinate. Valid examples: '40.7128', '-74.0060', '180', '-180', '0'."
-    if not regex.match(v):
-        raise ValueError(msg)
-
-    float_v = float(v)
-    if float_v > MAX_LONGITUDE or float_v < MIN_LONGITUDE:
-        raise ValueError(msg)
-
-    return v
-
-
-LatitudeCoordinate = Annotated[
-    str,
-    AfterValidator(validate_latitude),
-    Doc(
-        "A latitude coordinate, modeled as a string. "
-        "The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. "
-        "It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0."
-    ),
-]
-
-LongitudeCoordinate = Annotated[
-    str,
-    AfterValidator(validate_longitude),
-    Doc(
-        "A longitude coordinate, modeled as a string. "
-        "The coordinate must match the format conforming to the longitude "
-        "range of -180 to +180 degrees. It can be a floating-point number or an integer. "
-        "Valid examples: 40.7128, -74.0060, 180, -180, 0."
-    ),
-]
-
-
 class SiteBlockInactive(
     ProductBlockModel,
     lifecycle=[SubscriptionLifecycle.INITIAL],
@@ -85,7 +28,7 @@ class SiteBlockInactive(
 ):
     """A site that's currently inactive, see :class:`SiteBlock`."""
 
-    site_name: str | None = None
+    site_name: SiteName | None = None
     site_city: str | None = None
     site_country: str | None = None
     site_country_code: str | None = None
@@ -94,13 +37,13 @@ class SiteBlockInactive(
     site_internal_id: int | None = None
     site_bgp_community_id: int | None = None
     site_tier: SiteTier | None = None
-    site_ts_address: str | None = None
+    site_ts_address: IPAddress | None = None
 
 
 class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """A site that's currently being provisioned, see :class:`SiteBlock`."""
 
-    site_name: str
+    site_name: SiteName
     site_city: str
     site_country: str
     site_country_code: str
@@ -109,7 +52,7 @@ class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.
     site_internal_id: int
     site_bgp_community_id: int
     site_tier: SiteTier
-    site_ts_address: str
+    site_ts_address: IPAddress
 
 
 class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -117,7 +60,7 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
 
     #:  The name of the site, that will dictate part of the :term:`FQDN` of routers that are hosted at this site. For
     #:  example: ``router.X.Y.geant.net``, where X denotes the name of the site.
-    site_name: str
+    site_name: SiteName
     #:  The city at which the site is located.
     site_city: str
     #:  The country in which the site is located.
@@ -138,4 +81,4 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
     #:  The address of the terminal server that this router is connected to. The terminal server provides out of band
     #:  access. This is required in case a link goes down, or when a router is initially added to the network and it
     #:  does not have any IP trunks connected to it.
-    site_ts_address: str
+    site_ts_address: IPAddress

gso/products/product_blocks/super_pop_switch.py

@@ -8,7 +8,8 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import IPv4AddressType, PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import IPv4AddressType, PortNumber
 
 
 class SuperPopSwitchBlockInactive(

gso/products/product_blocks/switch.py

@@ -9,7 +9,8 @@ from gso.products.product_blocks.site import (
     SiteBlockInactive,
     SiteBlockProvisioning,
 )
-from gso.utils.shared_enums import PortNumber, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.ip_address import PortNumber
 
 
 class SwitchModel(strEnum):

gso/schedules/task_vacuum.py

@@ -7,7 +7,7 @@ from gso.worker import celery
 
 
 @celery.task
-@scheduler(CronScheduleConfig(name="Clean up tasks", hour="1"))
+@scheduler(CronScheduleConfig(name="Clean up tasks", hour="1", minute="0"))
 def vacuum_tasks() -> None:
     """Run all cleanup tasks every 1 AM UTC."""
     start_process("task_clean_up_tasks")

gso/services/infoblox.py

@@ -10,7 +10,7 @@ from infoblox_client.exceptions import (
 )
 
 from gso.settings import IPAMParams, load_oss_params
-from gso.utils.shared_enums import IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 
 logger = getLogger(__name__)
 NULL_MAC = "00:00:00:00:00:00"

gso/services/librenms_client.py

@@ -10,7 +10,7 @@ from requests import HTTPError, Response
 from requests.adapters import HTTPAdapter
 
 from gso.settings import load_oss_params
-from gso.utils.helpers import SNMPVersion
+from gso.utils.types.snmp import SNMPVersion
 
 logger = logging.getLogger(__name__)
 
@@ -40,7 +40,7 @@ class LibreNMSClient:
     ) -> Response:
         url = self.base_url + endpoint
         logger.debug("LibreNMS - Sending request", extra={"method": method, "endpoint": url, "form_data": data})
-        result = self.session.request(method, url, json=data, timeout=(0.5, 75))
+        result = self.session.request(method, url, json=data, timeout=(10, 75))
         logger.debug("LibreNMS - Received response", extra=result.__dict__)
 
         return result

gso/services/partners.py

 """A module that returns the partners available in :term:`GSO`."""
 
 from datetime import datetime
-from typing import Any
+from typing import Annotated, Any
 from uuid import uuid4
 
 from orchestrator.db import db
-from pydantic import BaseModel, ConfigDict, EmailStr, Field
+from pydantic import AfterValidator, BaseModel, ConfigDict, EmailStr, Field
 from sqlalchemy import func
 from sqlalchemy.exc import NoResultFound
 
 from gso.db.models import PartnerTable
 
 
+def validate_partner_name_unique(name: str) -> str:
+    """Validate that the name of a partner is unique."""
+    if filter_partners_by_name(name=name, case_sensitive=False):
+        msg = "Partner with this name already exists."
+        raise ValueError(msg)
+    return name
+
+
+def validate_partner_email_unique(email: EmailStr) -> EmailStr:
+    """Validate that the e-mail address of a partner is unique."""
+    email = email.lower()
+    if filter_partners_by_email(email=email, case_sensitive=False):
+        msg = "Partner with this email already exists."
+        raise ValueError(msg)
+    return email
+
+
+PartnerName = Annotated[str, AfterValidator(validate_partner_name_unique)]
+PartnerEmail = Annotated[EmailStr, AfterValidator(validate_partner_email_unique)]
+
+
 class PartnerSchema(BaseModel):
     """Partner schema."""
 
     partner_id: str = Field(default_factory=lambda: str(uuid4()))
-    name: str
-    email: EmailStr
+    name: PartnerName
+    email: PartnerEmail
 
     created_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
     updated_at: datetime = Field(default_factory=lambda: datetime.now().astimezone())
     model_config = ConfigDict(from_attributes=True)
 
 
+class ModifiedPartnerSchema(PartnerSchema):
+    """Partner schema when making a modification.
+
+    The name and email can be empty in this case, if they don't need changing.
+    """
+
+    name: PartnerName | None = None  # type: ignore[assignment]
+    email: PartnerEmail | None = None  # type: ignore[assignment]
+
+
 class PartnerNotFoundError(Exception):
     """Exception raised when a partner is not found."""
 
@@ -95,7 +126,7 @@ def create_partner(
 
 
 def edit_partner(
-    partner_data: PartnerSchema,
+    partner_data: ModifiedPartnerSchema,
 ) -> PartnerTable:
     """Edit an existing partner and update it in the database."""
     partner = get_partner_by_id(partner_id=partner_data.partner_id)

gso/services/subscriptions.py

@@ -69,20 +69,6 @@ def get_subscriptions(
     return [dict(zip(includes, result, strict=True)) for result in results]
 
 
-def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
-    """Retrieve active subscriptions specifically for sites.
-
-    :param includes: The fields to be included in the returned Subscription objects.
-    :type includes: list[str]
-
-    :return: A list of Subscription objects for sites.
-    :rtype: list[Subscription]
-    """
-    return get_subscriptions(
-        product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
-    )
-
-
 def get_router_subscriptions(
     includes: list[str] | None = None, lifecycles: list[SubscriptionLifecycle] | None = None
 ) -> list[SubscriptionType]:
@@ -122,6 +108,20 @@ def get_provisioning_router_subscriptions(includes: list[str] | None = None) ->
     )
 
 
+def get_active_switch_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+    """Retrieve active subscriptions specifically for switches.
+
+    :param includes: The fields to be included in the returned Subscription objects.
+    :type includes: list[str]
+
+    :return: A list of Subscription objects for switches.
+    :rtype: list[Subscription]
+    """
+    return get_subscriptions(
+        product_types=[ProductType.SWITCH], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+    )
+
+
 def get_active_iptrunk_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
     """Retrieve active subscriptions specifically for IP trunks.
 
@@ -233,6 +233,20 @@ def get_active_insync_subscriptions() -> list[SubscriptionTable]:
     )
 
 
+def get_active_site_subscriptions(includes: list[str] | None = None) -> list[SubscriptionType]:
+    """Retrieve active subscriptions specifically for sites.
+
+    :param includes: The fields to be included in the returned Subscription objects.
+    :type includes: list[str]
+
+    :return: A list of Subscription objects for sites.
+    :rtype: list[Subscription]
+    """
+    return get_subscriptions(
+        product_types=[ProductType.SITE], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=includes
+    )
+
+
 def get_site_by_name(site_name: str) -> Site:
     """Get a site by its name.
 

gso/settings.py

@@ -16,7 +16,7 @@ from pydantic_forms.types import UUIDstr, strEnum
 from pydantic_settings import BaseSettings
 from typing_extensions import Doc
 
-from gso.utils.shared_enums import PortNumber
+from gso.utils.types.ip_address import PortNumber
 
 logger = logging.getLogger(__name__)
 

gso/translations/en-GB.json

@@ -32,16 +32,18 @@
             "migrate_to_different_site": "Migrating to a different Site",
             "remove_configuration": "Remove configuration from the router",
             "clean_up_ipam": "Clean up related entries in IPAM",
-            "restore_isis_metric": "Restore ISIS metric to original value"
+            "restore_isis_metric": "Restore ISIS metric to original value",
+            "confirm_info": "Please verify this form looks correct."
         }
     },
     "workflow": {
         "activate_iptrunk": "Activate IP Trunk",
-        "activate_router": "Activate router",
-        "confirm_info": "Please verify this form looks correct.",
+        "activate_router": "Activate Router",
+        "activate_switch": "Activate Switch",
         "create_iptrunk": "Create IP Trunk",
         "create_router": "Create Router",
         "create_site": "Create Site",
+        "create_switch": "Create Switch",
         "deploy_twamp": "Deploy TWAMP",
         "migrate_iptrunk": "Migrate IP Trunk",
         "modify_isis_metric": "Modify the ISIS metric",
@@ -52,6 +54,7 @@
         "terminate_iptrunk": "Terminate IP Trunk",
         "terminate_router": "Terminate Router",
         "terminate_site": "Terminate Site",
+        "terminate_switch": "Terminate Switch",
         "redeploy_base_config": "Redeploy base config",
         "update_ibgp_mesh": "Update iBGP mesh",
         "create_imported_site": "NOT FOR HUMANS -- Import existing site",
@@ -67,7 +70,8 @@
         "import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
         "import_opengear": "NOT FOR HUMANS -- Finalize import into an OpenGear",
         "validate_iptrunk": "Validate IP Trunk configuration",
-        "validate_router": "Validate router configuration",
+        "validate_router": "Validate Router configuration",
+        "validate_switch": "Validate Switch configuration",
         "task_validate_geant_products": "Validation task for GEANT products",
         "task_send_email_notifications": "Send email notifications for failed tasks",
         "task_create_partners": "Create partner task",

gso/utils/device_info.py

@@ -32,6 +32,13 @@ class TierInfo:
             breakout_interfaces_per_slot=[36, 35, 34, 33],
             total_10g_interfaces=60,
         )
+        self.tier3 = ModuleInfo(
+            device_type="7750 SR2-se",
+            module_bays_slots=[1, 2],
+            module_type="XCMC-2SE-2",
+            breakout_interfaces_per_slot=[36, 35, 34, 33],
+            total_10g_interfaces=76,
+        )
 
     def get_module_by_name(self, name: str) -> ModuleInfo:
         """Retrieve a module by name."""

gso/utils/helpers.py

 """Helper methods that are used across :term:`GSO`."""
 
-import ipaddress
 import re
-from enum import StrEnum
+from typing import TYPE_CHECKING
 from uuid import UUID
 
-import pycountry
-from orchestrator.types import UUIDstr
-from pydantic import BaseModel, field_validator
-from pydantic_core.core_schema import ValidationInfo
 from pydantic_forms.validators import Choice
 
 from gso import settings
-from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock, PhysicalPortCapacity
 from gso.products.product_blocks.router import RouterRole
-from gso.products.product_blocks.site import LatitudeCoordinate, LongitudeCoordinate, SiteTier
 from gso.products.product_types.router import Router
+from gso.services import subscriptions
 from gso.services.netbox_client import NetboxClient
-from gso.services.subscriptions import get_active_router_subscriptions, get_active_subscriptions_by_field_and_value
-from gso.utils.shared_enums import IPv4AddressType, Vendor
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.utils.types.ip_address import IPv4AddressType
 
-
-class LAGMember(BaseModel):
-    """A :term:`LAG` member interface that consists of a name and description."""
-
-    interface_name: str
-    interface_description: str | None = None
-
-    def __hash__(self) -> int:
-        """Calculate the hash based on the interface name and description, so that uniqueness can be determined."""
-        return hash((self.interface_name, self.interface_description))
-
-
-class SNMPVersion(StrEnum):
-    """An enumerator for the two relevant versions of :term:`SNMP`: v2c and 3."""
-
-    V2C = "v2c"
-    V3 = "v3"
+if TYPE_CHECKING:
+    from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
 
 
 def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
@@ -57,7 +37,7 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
 def available_interfaces_choices_including_current_members(
     router_id: UUID,
     speed: str,
-    interfaces: list[IptrunkInterfaceBlock],
+    interfaces: list["IptrunkInterfaceBlock"],
 ) -> Choice | None:
     """Return a list of available interfaces for a given router and speed including the current members.
 
@@ -119,184 +99,6 @@ def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
     return f"49.51e5.0001.{re_split}.00"
 
 
-def validate_router_in_netbox(subscription_id: UUIDstr) -> UUIDstr:
-    """Verify if a device exists in Netbox.
-
-    Raises a :class:`ValueError` if the device is not found.
-
-    :param subscription_id: The :term:`UUID` of the router subscription.
-    :type subscription_id: :class:`UUIDstr`
-
-    :return: The :term:`UUID` of the router subscription.
-    :rtype: :class:`UUIDstr`
-    """
-    router_type = Router.from_subscription(subscription_id)
-    if router_type.router.vendor == Vendor.NOKIA:
-        device = NetboxClient().get_device_by_name(router_type.router.router_fqdn)
-        if not device:
-            msg = "The selected router does not exist in Netbox."
-            raise ValueError(msg)
-    return subscription_id
-
-
-def validate_iptrunk_unique_interface(interfaces: list[LAGMember]) -> list[LAGMember]:
-    """Verify if the interfaces are unique.
-
-    Raises a :class:`ValueError` if the interfaces are not unique.
-
-    :param interfaces: The list of interfaces.
-    :type interfaces: list[:class:`LAGMember`]
-
-    :return: The list of interfaces
-    :rtype: list[:class:`LAGMember`]
-    """
-    interface_names = [member.interface_name for member in interfaces]
-    if len(interface_names) != len(set(interface_names)):
-        msg = "Interfaces must be unique."
-        raise ValueError(msg)
-    return interfaces
-
-
-def validate_site_fields_is_unique(field_name: str, value: str | int) -> None:
-    """Validate that a site field is unique."""
-    if len(get_active_subscriptions_by_field_and_value(field_name, str(value))) > 0:
-        msg = f"{field_name} must be unique"
-        raise ValueError(msg)
-
-
-def validate_ipv4_or_ipv6(value: str) -> str:
-    """Validate that a value is a valid IPv4 or IPv6 address."""
-    try:
-        ipaddress.ip_address(value)
-    except ValueError as e:
-        msg = "Enter a valid IPv4 or IPv6 address."
-        raise ValueError(msg) from e
-    else:
-        return value
-
-
-def validate_country_code(country_code: str) -> str:
-    """Validate that a country code is valid."""
-    # Check for the UK code before attempting to look it up since it's known as "GB" in the pycountry database.
-    if country_code != "UK":
-        try:
-            pycountry.countries.lookup(country_code)
-        except LookupError as e:
-            msg = "Invalid or non-existent country code, it must be in ISO 3166-1 alpha-2 format."
-            raise ValueError(msg) from e
-    return country_code
-
-
-def validate_site_name(site_name: str) -> None:
-    """Validate the site name.
-
-    The site name must consist of three uppercase letters, optionally followed by a single digit.
-    """
-    pattern = re.compile(r"^[A-Z]{3}[0-9]?$")
-    if not pattern.match(site_name):
-        msg = (
-            "Enter a valid site name. It must consist of three uppercase letters (A-Z), followed by an optional single "
-            f"digit (0-9). Received: {site_name}"
-        )
-        raise ValueError(msg)
-
-
-class BaseSiteValidatorModel(BaseModel):
-    """A base site validator model extended by create site and by import site."""
-
-    site_bgp_community_id: int
-    site_internal_id: int
-    site_tier: SiteTier
-    site_ts_address: str
-    site_country_code: str
-    site_name: str
-    site_city: str
-    site_country: str
-    site_latitude: LatitudeCoordinate
-    site_longitude: LongitudeCoordinate
-    partner: str
-
-    @field_validator("site_ts_address")
-    def validate_ts_address(cls, site_ts_address: str) -> str:
-        """Validate that a terminal server address is valid."""
-        validate_ipv4_or_ipv6(site_ts_address)
-        return site_ts_address
-
-    @field_validator("site_country_code")
-    def country_code_must_exist(cls, country_code: str) -> str:
-        """Validate that the country code exists."""
-        validate_country_code(country_code)
-        return country_code
-
-    @field_validator("site_ts_address", "site_internal_id", "site_bgp_community_id", "site_name")
-    def field_must_be_unique(cls, value: str | int, info: ValidationInfo) -> str | int:
-        """Validate that a field is unique."""
-        if not info.field_name:
-            msg = "Field name must be provided."
-            raise ValueError(msg)
-
-        validate_site_fields_is_unique(info.field_name, value)
-
-        return value
-
-    @field_validator("site_name")
-    def site_name_must_be_valid(cls, site_name: str) -> str:
-        """Validate the site name.
-
-        The site name must consist of three uppercase letters, followed by an optional single digit.
-        """
-        validate_site_name(site_name)
-        return site_name
-
-
-def validate_interface_name_list(interface_name_list: list, vendor: str) -> list:
-    """Validate that the provided interface name matches the expected pattern.
-
-    The expected pattern for the interface name is one of 'ge', 'et', 'xe' followed by a dash '-',
-    then a number between 0 and 19, a forward slash '/', another number between 0 and 99,
-    another forward slash '/', and ends with a number between 0 and 99.
-    For example: 'xe-1/0/0'.
-
-    :param list interface_name_list: List of interface names to validate.
-    :param str vendor: Router vendor to check interface names
-
-    :return list: The list of interface names if all match was successful, otherwise it will throw a ValueError
-                  exception.
-    """
-    # For Nokia nothing to do
-    if vendor == Vendor.NOKIA:
-        return interface_name_list
-    pattern = re.compile(r"^(ge|et|xe)-1?[0-9]/[0-9]{1,2}/[0-9]{1,2}$")
-    for interface in interface_name_list:
-        if not bool(pattern.match(interface.interface_name)):
-            error_msg = (
-                f"Invalid interface name. The interface name should be of format: xe-1/0/0. "
-                f"Got: [{interface.interface_name}]"
-            )
-            raise ValueError(error_msg)
-    return interface_name_list
-
-
-def validate_tt_number(tt_number: str) -> str:
-    """Validate a string to match a specific pattern.
-
-    This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
-
-    :param str tt_number: The TT number as string to validate
-
-    :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
-    """
-    pattern = r"^TT#\d{16}$"
-    if not bool(re.match(pattern, tt_number)):
-        err_msg = (
-            f"The given TT number: {tt_number} is not valid. "
-            f" A valid TT number starts with 'TT#' followed by 16 digits."
-        )
-        raise ValueError(err_msg)
-
-    return tt_number
-
-
 def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
     """Generate an :term:`FQDN` from a hostname, site name, and a country code."""
     oss = settings.load_oss_params()
@@ -318,7 +120,9 @@ def generate_inventory_for_active_routers(
     :return: A dictionary representing the inventory of active routers.
     :rtype: dict[str, Any]
     """
-    all_routers = [Router.from_subscription(r["subscription_id"]) for r in get_active_router_subscriptions()]
+    all_routers = [
+        Router.from_subscription(r["subscription_id"]) for r in subscriptions.get_active_router_subscriptions()
+    ]
     exclude_routers = exclude_routers or []
 
     return {
@@ -351,3 +155,33 @@ def calculate_recommended_minimum_links(iptrunk_number_of_members: int, iptrunk_
     if iptrunk_speed == PhysicalPortCapacity.FOUR_HUNDRED_GIGABIT_PER_SECOND:
         return iptrunk_number_of_members - 1
     return iptrunk_number_of_members
+
+
+def active_site_selector() -> Choice:
+    """Generate a dropdown selector for choosing an active site in an input form."""
+    site_subscriptions = {
+        str(site["subscription_id"]): site["description"]
+        for site in subscriptions.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]
+
+
+def active_router_selector() -> Choice:
+    """Generate a dropdown selector for choosing an active Router in an input form."""
+    router_subscriptions = {
+        str(router["subscription_id"]): router["description"]
+        for router in subscriptions.get_active_router_subscriptions(includes=["subscription_id", "description"])
+    }
+
+    return Choice("Select a router", zip(router_subscriptions.keys(), router_subscriptions.items(), strict=True))  # type: ignore[arg-type]
+
+
+def active_switch_selector() -> Choice:
+    """Generate a dropdown selector for choosing an active Switch in an input form."""
+    switch_subscriptions = {
+        str(switch["subscription_id"]): switch["description"]
+        for switch in subscriptions.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]