Fixed linting errors.

gso/api/api_v1/api.py

@@ -2,12 +2,9 @@
 
 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 imports
 
 api_router = APIRouter()
-api_router.include_router(
-    imports.router, prefix="/imports", dependencies=[Depends(opa_security_default)]
-)
\ No newline at end of file
+api_router.include_router(imports.router, prefix="/imports", dependencies=[Depends(opa_security_default)])

gso/api/api_v1/endpoints/imports.py

 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}

gso/cli/import_sites.py

@@ -4,7 +4,7 @@ app: typer.Typer = typer.Typer()
 
 
 @app.command()
-def import_sites():
+def import_sites() -> None:
     """Import sites from a source."""
     # TODO: Implement this CLI command to import sites from a source.
     typer.echo("Importing sites...")

gso/services/crm.py

-from typing import Optional
+from typing import Any, Dict
 
 
 class CustomerNotFoundError(Exception):
     """Exception raised when a customer is not found."""
+
     pass
 
 
@@ -15,7 +16,7 @@ def all_customers() -> list[dict]:
     ]
 
 
-def get_customer_by_name(name: str) -> Optional[dict]:
+def get_customer_by_name(name: str) -> Dict[str, Any]:
     for customer in all_customers():
         if customer["name"] == name:
             return customer

gso/workflows/tasks/import_router.py

 import ipaddress
 from typing import Optional
+from uuid import UUID
 
 from orchestrator import workflow
 from orchestrator.db import (
     ProductTable,
-    SubscriptionTable,
-    SubscriptionInstanceValueTable,
-    SubscriptionInstanceTable,
     ResourceTypeTable,
+    SubscriptionInstanceTable,
+    SubscriptionInstanceValueTable,
+    SubscriptionTable,
 )
-from orchestrator.domain.base import S
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target
-from orchestrator.types import FormGenerator, State, UUIDstr, SubscriptionLifecycle
-from orchestrator.workflow import StepList, init, step, done
-from orchestrator.workflows.steps import store_process_subscription, set_status, resync
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
+from orchestrator.workflow import StepList, done, init, step
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 
-from gso.products import Site
 from gso.products.product_blocks import router as router_pb
-from gso.products.product_blocks.router import RouterVendor, RouterRole
+from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_types import router
 from gso.products.product_types.router import RouterInactive
+from gso.products.product_types.site import Site
 from gso.services.crm import get_customer_by_name
 
 
 @step("Create subscription")
 def create_subscription(customer: str) -> State:
-    customer_id: UUIDstr = get_customer_by_name(customer)["id"]
-    product_id: UUIDstr = ProductTable.query.filter_by(name="Router").first().product_id
+    customer_id: UUID = get_customer_by_name(customer)["id"]
+    product_id: UUID = ProductTable.query.filter_by(name="Router").first().product_id
     subscription = RouterInactive.from_product_id(product_id, customer_id)
 
     return {
@@ -61,7 +61,7 @@ def initial_input_form_generator() -> FormGenerator:
     return user_input.dict()
 
 
-def get_site_by_name(site_name: str) -> S:
+def get_site_by_name(site_name: str) -> Site:
     subscription = (
         SubscriptionTable.query.join(
             ProductTable, SubscriptionInstanceTable, SubscriptionInstanceValueTable, ResourceTypeTable

gso/workflows/tasks/import_site.py

-from uuid import uuid4
+from uuid import UUID, 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.types import FormGenerator, State, SubscriptionLifecycle
 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
@@ -50,15 +49,14 @@ def generate_initial_input_form() -> FormGenerator:
     initial_input_form=generate_initial_input_form,
 )
 def import_site() -> StepList:
-    """
-    Workflow to import a site without provisioning it.
-    """
+    """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
+        init
+        >> create_subscription
+        >> store_process_subscription(Target.CREATE)
+        >> initialize_subscription
+        >> set_status(SubscriptionLifecycle.ACTIVE)
+        >> resync
+        >> done
     )