Skip to content
Snippets Groups Projects
Commit 6ab32a16 authored by Saket Agrahari's avatar Saket Agrahari
Browse files

update to add output to file for import_edge_port

parent 2df16921
Branches
No related tags found
1 merge request!421Draft: update to add output to file for import_edge_port
Pipeline #94037 failed
...@@ -10,10 +10,6 @@ from typing import Self, TypeVar ...@@ -10,10 +10,6 @@ from typing import Self, TypeVar
import typer import typer
import yaml import yaml
from orchestrator.db import db
from orchestrator.services.processes import start_process
from orchestrator.services.products import get_product_by_id
from orchestrator.types import SubscriptionLifecycle
from pydantic import BaseModel, NonNegativeInt, ValidationError, field_validator, model_validator from pydantic import BaseModel, NonNegativeInt, ValidationError, field_validator, model_validator
from pydantic_forms.types import UUIDstr from pydantic_forms.types import UUIDstr
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
...@@ -55,6 +51,10 @@ from gso.utils.types.ip_address import ( ...@@ -55,6 +51,10 @@ from gso.utils.types.ip_address import (
) )
from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
from gso.workflows.l3_core_service.shared import L3_CREAT_IMPORTED_WF_MAP, L3_IMPORT_WF_MAP, L3ProductNameType from gso.workflows.l3_core_service.shared import L3_CREAT_IMPORTED_WF_MAP, L3_IMPORT_WF_MAP, L3ProductNameType
from orchestrator.db import db
from orchestrator.services.processes import start_process
from orchestrator.services.products import get_product_by_id
from orchestrator.types import SubscriptionLifecycle
app: typer.Typer = typer.Typer() app: typer.Typer = typer.Typer()
IMPORT_WAIT_MESSAGE = "Waiting for the dust to settle before importing new products..." IMPORT_WAIT_MESSAGE = "Waiting for the dust to settle before importing new products..."
...@@ -540,18 +540,32 @@ def import_opengear(filepath: str = common_filepath_option) -> None: ...@@ -540,18 +540,32 @@ def import_opengear(filepath: str = common_filepath_option) -> None:
@app.command() @app.command()
def import_edge_port(filepath: str = common_filepath_option) -> None: def import_edge_port(filepath: str, output_dir: str | None = None) -> None:
"""Import Edge Port into GSO.""" """Import Edge Port into GSO."""
successfully_imported_data = [] successfully_imported_data = []
data = _read_data(Path(filepath)) data = _read_data(Path(filepath))
# Save successfully imported data to a file if output_dir is provided
if output_dir:
output_file = Path(output_dir) / "successful_imported_ep.json"
output_file.parent.mkdir(parents=True, exist_ok=True) # Ensure the directory exists
# Load existing data if the file exists
if output_file.exists():
with output_file.open("r", encoding="utf-8") as f:
existing_data = json.load(f)
else:
existing_data = {}
for edge_port in data: for edge_port in data:
typer.echo(f"Importing Edge Port {edge_port["name"]} on {edge_port["node"]}. ") typer.echo(f"Importing Edge Port {edge_port["name"]} on {edge_port["node"]}. ")
try: try:
key = f"{edge_port["name"]}@{edge_port["node"]}"
edge_port["node"] = _get_router_subscription_id(edge_port["node"]) edge_port["node"] = _get_router_subscription_id(edge_port["node"])
initial_data = EdgePortImportModel(**edge_port) initial_data = EdgePortImportModel(**edge_port)
start_process("create_imported_edge_port", [initial_data.model_dump()]) start_process("create_imported_edge_port", [initial_data.model_dump()])
successfully_imported_data.append(edge_port["name"]) successfully_imported_data.append(edge_port["name"])
typer.echo(f"Successfully imported Edge Port {edge_port["name"]} on {edge_port["node"]}.") typer.echo(f"Successfully imported Edge Port {edge_port["name"]} on {edge_port["node"]}.")
existing_data[key] = edge_port["node"]
except ValidationError as e: except ValidationError as e:
typer.echo(f"Validation error: {e}") typer.echo(f"Validation error: {e}")
...@@ -570,6 +584,11 @@ def import_edge_port(filepath: str = common_filepath_option) -> None: ...@@ -570,6 +584,11 @@ def import_edge_port(filepath: str = common_filepath_option) -> None:
for item in successfully_imported_data: for item in successfully_imported_data:
typer.echo(f"- {item}") typer.echo(f"- {item}")
# Write the updated map back to the file
with output_file.open("w", encoding="utf-8") as f:
json.dump(existing_data, f, indent=4)
typer.echo(f"Successfully imported edge ports saved to {output_file}")
@app.command() @app.command()
def import_iptrunks(filepath: str = common_filepath_option) -> None: def import_iptrunks(filepath: str = common_filepath_option) -> None:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment