docs/source/module/utils/types/unique_field.rst

+``gso.utils.types.unique_field``
+================================
+
+.. automodule:: gso.utils.types.unique_field
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/create_edge_port.rst

+``gso.workflows.edge_port.create_edge_port``
+============================================
+
+.. automodule:: gso.workflows.edge_port.create_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/create_imported_edge_port.rst

+``gso.workflows.edge_port.create_imported_edge_port``
+=====================================================
+
+.. automodule:: gso.workflows.edge_port.create_imported_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/import_edge_port.rst

+``gso.workflows.edge_port.import_edge_port``
+============================================
+
+.. automodule:: gso.workflows.edge_port.import_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/index.rst

+``gso.workflows.edge_port``
+===========================
+
+.. automodule:: gso.workflows.edge_port
+   :members:
+   :show-inheritance:
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 2
+   :titlesonly:
+
+   create_edge_port
+   modify_edge_port
+   terminate_edge_port
+   validate_edge_port
+   import_edge_port
+   create_imported_edge_port

docs/source/module/workflows/edge_port/modify_edge_port.rst

+``gso.workflows.edge_port.modify_edge_port``
+============================================
+
+.. automodule:: gso.workflows.edge_port.modify_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/terminate_edge_port.rst

+``gso.workflows.edge_port.terminate_edge_port``
+===============================================
+
+.. automodule:: gso.workflows.edge_port.terminate_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/edge_port/validate_edge_port.rst

+``gso.workflows.edge_port.validate_edge_port``
+==============================================
+
+.. automodule:: gso.workflows.edge_port.validate_edge_port
+   :members:
+   :show-inheritance:

docs/source/module/workflows/geant_ip/create_geant_ip.rst

+``gso.workflows.geant_ip.create_geant_ip``
+==========================================
+
+.. automodule:: gso.workflows.geant_ip.create_geant_ip
+   :members:
+   :show-inheritance:

docs/source/module/workflows/geant_ip/create_imported_geant_ip.rst

+``gso.workflows.geant_ip.create_imported_geant_ip``
+====================================================
+
+.. automodule:: gso.workflows.geant_ip.create_imported_geant_ip
+   :members:
+   :show-inheritance:

docs/source/module/workflows/geant_ip/import_geant_ip.rst

+``gso.workflows.geant_ip.import_geant_ip``
+==========================================
+
+.. automodule:: gso.workflows.geant_ip.import_geant_ip
+   :members:
+   :show-inheritance:

docs/source/module/workflows/geant_ip/index.rst

+``gso.workflows.geant_ip``
+==========================
+
+.. automodule:: gso.workflows.geant_ip
+   :members:
+   :show-inheritance:
+
+Submodules
+----------
+
+.. toctree::
+   :maxdepth: 2
+   :titlesonly:
+
+   create_geant_ip
+   create_imported_geant_ip
+   import_geant_ip
+   migrate_geant_ip
+   modify_geant_ip

docs/source/module/workflows/geant_ip/migrate_geant_ip.rst

+``gso.workflows.geant_ip.migrate_geant_ip``
+===========================================
+
+.. automodule:: gso.workflows.geant_ip.migrate_geant_ip
+   :members:
+   :show-inheritance:

docs/source/module/workflows/geant_ip/modify_geant_ip.rst

+``gso.workflows.geant_ip.modify_geant_ip``
+==========================================
+
+.. automodule:: gso.workflows.geant_ip.modify_geant_ip
+   :members:
+   :show-inheritance:

docs/source/module/workflows/index.rst

@@ -19,3 +19,5 @@ Subpackages
    site/index
    super_pop_switch/index
    tasks/index
+   edge_port/index
+   geant_ip/index
\ No newline at end of file

docs/vale/.vale.ini

@@ -25,6 +25,7 @@ custom.Contractions = YES
 ; Using a "regular" - instead of an en dash is totally fine
 Microsoft.Negative = NO
 Microsoft.RangeFormat = NO
+Microsoft.We = suggestion
 
 TokenIgnores = (:term:`\S+`), (:param \S+(?: \S+)?:), (:type \S+:), (:return \S+:), (:rtype: \S+), (:class:`\S+`)
 

docs/vale/styles/config/vocabularies/geant-jargon/accept.txt

@@ -31,3 +31,7 @@ OPA
 OIDC
 HTTPBearer
 Kentik
+UTC
+EARL
+SURF
+[Ee]nsure

docs/vale/styles/custom/Contractions.yml

@@ -7,7 +7,6 @@ swap:
   can't: cannot
   couldn't: could not
   didn't: did not
-  don't: do not
   doesn't: does not
   hasn't: has not
   haven't: have not

gso/cli/imports.py

@@ -18,8 +18,11 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from gso.db.models import PartnerTable
 from gso.products import ProductType
+from gso.products.product_blocks.bgp_session import IPFamily
+from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_blocks.iptrunk import IptrunkType
 from gso.products.product_blocks.router import RouterRole
+from gso.products.product_blocks.service_binding_port import VLAN_ID
 from gso.services.partners import (
     PartnerEmail,
     PartnerName,
@@ -31,10 +34,10 @@ from gso.services.subscriptions import (
     get_active_subscriptions_by_field_and_value,
     get_subscriptions,
 )
-from gso.utils.shared_enums import Vendor
+from gso.utils.shared_enums import SBPType, 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
+from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv6AddressType, PortNumber
 
 app: typer.Typer = typer.Typer()
 
@@ -161,6 +164,95 @@ class OpenGearImportModel(BaseModel):
     opengear_wan_gateway: IPv4AddressType
 
 
+class EdgePortImportModel(BaseModel):
+    """Required fields for importing an existing :class:`gso.products.product_types.edge_port`."""
+
+    node: str
+    service_type: EdgePortType
+    speed: PhysicalPortCapacity
+    encapsulation: EncapsulationType
+    name: str
+    minimum_links: int
+    geant_ga_id: str | None
+    mac_address: str | None
+    partner: str
+    enable_lacp: bool
+    ignore_if_down: bool
+    ae_members: LAGMemberList[LAGMember]
+    description: str | None = None
+
+    @field_validator("partner")
+    def check_if_partner_exists(cls, value: str) -> str:
+        """Validate that the partner exists."""
+        try:
+            get_partner_by_name(value)
+        except PartnerNotFoundError as e:
+            msg = f"Partner {value} not found"
+            raise ValueError(msg) from e
+
+        return value
+
+    @field_validator("node")
+    def validate_node(cls, value: str) -> str:
+        """Check if the node is an active PE router in :term:`GSO`."""
+        pe_routers = {
+            str(router.subscription_id)
+            for router in get_active_subscriptions_by_field_and_value("router_role", RouterRole.PE)
+        }
+        if value not in pe_routers:
+            msg = f"Router {value} not found"
+            raise ValueError(msg)
+
+        return value
+
+    @model_validator(mode="after")
+    def check_members(self) -> Self:
+        """Amount of :term:`LAG` members has to match and meet the minimum requirement."""
+        if len(self.ae_members) < self.minimum_links:
+            msg = f"Number of members should be at least {self.minimum_links} (edge_port_minimum_links)"
+            raise ValueError(msg)
+        return self
+
+
+class GeantIPImportModel(BaseModel):
+    """Import GEANT IP model."""
+
+    partner: str
+    service_binding_ports: list["GeantIPImportModel.ServiceBindingPort"]
+
+    class BaseBGPPeer(BaseModel):
+        """Base BGP Peer model."""
+
+        bfd_enabled: bool = False
+        bfd_interval: int | None = None
+        bfd_multiplier: int | None = None
+        has_custom_policies: bool = False
+        authentication_key: str
+        multipath_enabled: bool = False
+        send_default_route: bool = False
+        is_passive: bool = False
+        peer_address: IPAddress
+        families: list[IPFamily]
+        is_multi_hop: bool
+        rtbh_enabled: bool  # whether Remote Triggered Blackhole is enabled
+
+    class ServiceBindingPort(BaseModel):
+        """Service Binding model."""
+
+        edge_port: str
+        ap_type: str
+        geant_sid: str
+        sbp_type: SBPType = SBPType.L3
+        is_tagged: bool = False
+        vlan_id: VLAN_ID
+        custom_firewall_filters: bool = False
+        ipv4_address: IPv4AddressType
+        ipv6_address: IPv6AddressType
+        rtbh_enabled: bool = True
+        is_multi_hop: bool = True
+        bgp_peers: list["GeantIPImportModel.BaseBGPPeer"]
+
+
 T = TypeVar(
     "T",
     SiteImportModel,
@@ -169,6 +261,8 @@ T = TypeVar(
     SuperPopSwitchImportModel,
     OfficeRouterImportModel,
     OpenGearImportModel,
+    EdgePortImportModel,
+    GeantIPImportModel,
 )
 
 common_filepath_option = typer.Option(
@@ -219,7 +313,7 @@ def _generic_import_product(
     successfully_imported_data = []
     data = _read_data(file_path)
     for details in data:
-        details["partner"] = "GEANT"
+        details["partner"] = details.get("partner", "GEANT")
         typer.echo(f"Creating imported {name_key}: {details[name_key]}")
         try:
             initial_data = import_model(**details)
@@ -297,6 +391,38 @@ def import_opengear(filepath: str = common_filepath_option) -> None:
     )
 
 
+@app.command()
+def import_edge_port(filepath: str = common_filepath_option) -> None:
+    """Import Edge Port into GSO."""
+    successfully_imported_data = []
+    data = _read_data(Path(filepath))
+    for edge_port in data:
+        typer.echo(f"Importing Edge Port {edge_port["name"]} on {edge_port["node"]}. ")
+        try:
+            edge_port["node"] = _get_router_subscription_id(edge_port["node"])
+            initial_data = EdgePortImportModel(**edge_port)
+            start_process("create_imported_edge_port", [initial_data.model_dump()])
+            successfully_imported_data.append(edge_port["name"])
+            typer.echo(f"Successfully imported Edge Port {edge_port["name"]} on {edge_port["node"]}.")
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")
+
+    typer.echo("Waiting for the dust to settle before moving on the importing new products...")
+    time.sleep(1)
+
+    edge_port_ids = get_subscriptions(
+        [ProductType.IMPORTED_EDGE_PORT], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=["subscription_id"]
+    )
+    for subscription_id in edge_port_ids:
+        typer.echo(f"Migrating Edge Port {subscription_id}")
+        start_process("import_edge_port", [subscription_id])
+
+    if successfully_imported_data:
+        typer.echo("Successfully imported Edge Ports:")
+        for item in successfully_imported_data:
+            typer.echo(f"- {item}")
+
+
 @app.command()
 def import_iptrunks(filepath: str = common_filepath_option) -> None:
     """Import IP trunks into GSO."""
@@ -390,3 +516,40 @@ def import_partners(file_path: str = typer.Argument(..., help="Path to the CSV f
         typer.echo(f"Failed to import partners: {e}")
     finally:
         db.session.close()
+
+
+@app.command()
+def import_geant_ip(filepath: str = common_filepath_option) -> None:
+    """Import GEANT IP into GSO."""
+    successfully_imported_data = []
+    geant_ip_list = _read_data(Path(filepath))
+
+    for geant_ip in geant_ip_list:
+        partner = geant_ip["partner"]
+        typer.echo(f"Creating imported GEANT IP for {partner}")
+
+        try:
+            initial_data = GeantIPImportModel(**geant_ip)
+            start_process("create_imported_geant_ip", [initial_data.model_dump()])
+            edge_ports = [sbp["edge_port"] for sbp in geant_ip["service_binding_ports"]]
+            successfully_imported_data.append(edge_ports)
+            typer.echo(f"Successfully created imported GEANT IP for {partner}")
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")
+
+    typer.echo("Waiting for the dust to settle before importing new products...")
+    time.sleep(1)
+
+    # Migrate new products from imported to "full" counterpart.
+    imported_products = get_subscriptions(
+        [ProductType.IMPORTED_GEANT_IP], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=["subscription_id"]
+    )
+
+    for subscription_id in imported_products:
+        typer.echo(f"Importing {subscription_id}")
+        start_process("import_geant_ip", [subscription_id])
+
+    if successfully_imported_data:
+        typer.echo("Successfully created imported GEANT IPs:")
+        for item in successfully_imported_data:
+            typer.echo(f"- {item}")