Added import site ability.

gso/__init__.py

+from typer import Typer
+
+
+def load_gso_cli(app: Typer) -> None:
+    from gso.cli import import_sites
+
+    app.add_typer(import_sites.app, name="import_sites")

gso/api/api_v1/api.py

+"""Module that implements process related API endpoints."""
+
+from fastapi.param_functions import Depends
+from fastapi.routing import APIRouter
+
+from orchestrator.security import opa_security_default
+
+from gso.api.api_v1.endpoints import import_site
+
+api_router = APIRouter()
+api_router.include_router(
+    import_site.router, prefix="/import/site", dependencies=[Depends(opa_security_default)]
+)
\ No newline at end of file

gso/api/api_v1/endpoints/import_site.py

+from typing import Dict, Any
+
+from fastapi import HTTPException
+from fastapi.routing import APIRouter
+from orchestrator.services import processes
+from pydantic import BaseModel
+
+from gso.products.product_blocks.site import SiteTier
+
+router = APIRouter()
+
+
+class SiteImport(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
+
+
+@router.post("/", tags=["Import"])
+def import_site(site: SiteImport) -> Dict[str, Any]:
+    """
+    Import site by running the import_site workflow.
+    """
+    pid = processes.start_process("import_site", [site.dict()])
+    if pid is None:
+        raise HTTPException(status_code=500, detail="Failed to start the process.")
+
+    process = processes._get_process(pid)  # pylint: disable=protected-access
+    if process.last_status == "failed":
+        raise HTTPException(
+            status_code=500, detail=f"Process {pid} failed because of an internal error.")
+    return {"pid": str(pid)}

gso/cli/import_sites.py

+from typing import Any, Dict, Generator
+
+import requests
+import typer
+from pydantic import ValidationError
+
+from gso.api.api_v1.endpoints.import_site import SiteImport, import_site
+from gso.products.product_blocks.site import SiteTier
+
+app: typer.Typer = typer.Typer()
+
+
+def get_site_details() -> Generator[Dict[str, Any], None, None]:
+    site_list_url = "https://prod-inventory-provider01.geant.org/neteng/pops"
+    site_details_url_template = "https://prod-inventory-provider01.geant.org/neteng/pop/{site}"
+    site_list_response = requests.get(site_list_url)
+    site_list = site_list_response.json()
+
+    for site in site_list:
+        site_details_url = site_details_url_template.format(site=site)
+        site_details_response = requests.get(site_details_url)
+        yield site_details_response.json()
+
+
+@app.command()
+def import_from_inventory_provider():
+    """
+    Import sites into GSO using Inventory Provider API.
+    """
+    for site_details in get_site_details():
+        try:
+            mapped_site_details = {
+                "site_name": site_details["name"],
+                "site_city": site_details["city"],
+                "site_country": site_details["country"],
+                "site_latitude": site_details["latitude"],
+                "site_longitude": site_details["longitude"],
+                "site_internal_id": 0,
+                "site_country_code": "NL",
+                "site_bgp_community_id": 0,
+                "site_tier": SiteTier.tier1
+            }
+            initial_data = SiteImport(**mapped_site_details)
+            import_site(initial_data)
+            typer.echo(f"Successfully imported site: {initial_data.site_name}")
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")

gso/main.py

 """The main module that runs {term}`GSO`."""
+import typer
 from orchestrator import OrchestratorCore
 from orchestrator.cli.main import app as core_cli
 from orchestrator.settings import AppSettings
 
 import gso.products  # noqa: F401
 import gso.workflows  # noqa: F401
+from gso import load_gso_cli
+from gso.api.api_v1.api import api_router
 
 app = OrchestratorCore(base_settings=AppSettings())
+app.include_router(api_router, prefix="/api")
+
+
+def init_cli_app() -> typer.Typer:
+    load_gso_cli(core_cli)
+    return core_cli()
+
 
 if __name__ == "__main__":
-    core_cli()
+    init_cli_app()

gso/workflows/__init__.py

@@ -8,3 +8,4 @@ LazyWorkflowInstance("gso.workflows.iptrunk.modify_trunk_interface", "modify_tru
 LazyWorkflowInstance("gso.workflows.iptrunk.terminate_iptrunk", "terminate_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric", "modify_isis_metric")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
+LazyWorkflowInstance("gso.workflows.tasks.import_site", "import_site")
\ No newline at end of file

gso/workflows/tasks/import_site.py

+from uuid import uuid4
+
+from orchestrator.db.models import ProductTable
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import State, SubscriptionLifecycle, FormGenerator
+from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic.types import UUID
+
+from gso.products.product_blocks.site import SiteTier
+from gso.products.product_types import site
+from gso.workflows.site.create_site import initialize_subscription
+
+
+@step("Create subscription")
+def create_subscription() -> State:
+    product_id: UUID = ProductTable.query.filter_by(product_type="Site").first().product_id
+    subscription = site.SiteInactive.from_product_id(product_id, uuid4())
+
+    return {
+        "subscription": subscription,
+        "subscription_id": subscription.subscription_id,
+    }
+
+
+def generate_initial_input_form() -> FormGenerator:
+    class ImportSite(FormPage):
+        class Config:
+            title = "Import Site"
+
+        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
+
+    user_input = yield ImportSite
+    return user_input.dict()
+
+
+@workflow(
+    "Import Site",
+    target=Target.SYSTEM,
+    initial_input_form=generate_initial_input_form,
+)
+def import_site() -> StepList:
+    """
+    Workflow to import a site without provisioning it.
+    """
+    return (
+            init
+            >> create_subscription
+            >> store_process_subscription(Target.CREATE)
+            >> initialize_subscription
+            >> set_status(SubscriptionLifecycle.ACTIVE)
+            >> resync
+            >> done
+    )

requirements.txt

@@ -11,3 +11,10 @@ mypy
 ruff
 sphinx
 sphinx-rtd-theme
+requests
+typer
+fastapi
+typer
+alembic
+SQLAlchemy
+setuptools
\ No newline at end of file