Skip to content
Snippets Groups Projects

Feature/nat 217 import sites

Merged Neda Moeini requested to merge feature/NAT-217-Import-Sites into develop
6 files
+ 49
47
Compare changes
  • Side-by-side
  • Inline
Files
6
import ipaddress
from typing import Dict, Any, Optional
from typing import Any, Dict, Optional
from uuid import UUID
from fastapi import HTTPException, status
@@ -15,9 +15,9 @@ router = APIRouter()
def start_process(process_name: str, data: dict) -> UUID:
"""Utility function to start a process and handle common exceptions."""
"""Start a process and handle common exceptions."""
pid = processes.start_process(process_name, [data])
pid: UUID = processes.start_process(process_name, [data])
if pid is None:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to start the process.")
@@ -46,16 +46,19 @@ class SiteImport(BaseModel):
@router.post("/sites", status_code=status.HTTP_201_CREATED, tags=["Import"])
def import_site(site: SiteImport) -> Dict[str, Any]:
"""
Import site by running the import_site workflow.
"""Import site by running the import_site workflow.
Args:
- site: A SiteImport object containing site details.
----
site (SiteImport): The site information to be imported.
Returns:
- A dictionary containing the detail message and the process id.
-------
dict: A dictionary containing the process id of the started process and detail message.
Raises:
- HTTPException: If there's an error in the process or if the site already exists.
------
HTTPException: If the site already exists or if there's an error in the process.
"""
try:
subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
@@ -88,18 +91,21 @@ class RouterImportModel(BaseModel):
@router.post("/routers", status_code=status.HTTP_201_CREATED, tags=["Import"])
def import_router(router_data: RouterImportModel):
"""
Import router by running the import_router workflow.
def import_router(router_data: RouterImportModel) -> Dict[str, Any]:
"""Import a router by running the import_router workflow.
Args:
- router_data: A RouterImportModel object containing router details.
----
router_data (RouterImportModel): The router information to be imported.
Returns:
- A dictionary containing the detail message and the process id.
-------
dict: A dictionary containing the process id of the started process and detail message.
Raises:
- HTTPException: If there's an error in the process.
------
HTTPException: If there's an error in the process.
"""
pid = start_process("import_router", router_data.dict())
return {"detail": f"Router added successfully", "pid": pid}
return {"detail": "Router added successfully", "pid": pid}
Loading