From 05269a8b65d54853c039de60547b154849870ac8 Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@GA0479-NMOEINI.local>
Date: Fri, 18 Aug 2023 14:50:17 +0200
Subject: [PATCH] Fixed linting errors.
---
gso/api/api_v1/api.py | 5 +---
gso/api/api_v1/endpoints/imports.py | 36 ++++++++++++++++------------
gso/cli/import_sites.py | 2 +-
gso/services/crm.py | 5 ++--
gso/workflows/tasks/import_router.py | 24 +++++++++----------
gso/workflows/tasks/import_site.py | 24 +++++++++----------
6 files changed, 49 insertions(+), 47 deletions(-)
diff --git a/gso/api/api_v1/api.py b/gso/api/api_v1/api.py
index 8166a11f..b7ba2d52 100644
--- a/gso/api/api_v1/api.py
+++ b/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)])
diff --git a/gso/api/api_v1/endpoints/imports.py b/gso/api/api_v1/endpoints/imports.py
index fe510fea..48da3fd2 100644
--- a/gso/api/api_v1/endpoints/imports.py
+++ b/gso/api/api_v1/endpoints/imports.py
@@ -1,5 +1,5 @@
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}
diff --git a/gso/cli/import_sites.py b/gso/cli/import_sites.py
index 4a4af843..36e0cb49 100644
--- a/gso/cli/import_sites.py
+++ b/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...")
diff --git a/gso/services/crm.py b/gso/services/crm.py
index cdff901f..92fb150d 100644
--- a/gso/services/crm.py
+++ b/gso/services/crm.py
@@ -1,8 +1,9 @@
-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
diff --git a/gso/workflows/tasks/import_router.py b/gso/workflows/tasks/import_router.py
index c6cd8d06..5b0468b6 100644
--- a/gso/workflows/tasks/import_router.py
+++ b/gso/workflows/tasks/import_router.py
@@ -1,33 +1,33 @@
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
diff --git a/gso/workflows/tasks/import_site.py b/gso/workflows/tasks/import_site.py
index 2dbc2a8a..d743628f 100644
--- a/gso/workflows/tasks/import_site.py
+++ b/gso/workflows/tasks/import_site.py
@@ -1,12 +1,11 @@
-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
)
--
GitLab