Commit 443b6289 authored by Adrian Pino's avatar Adrian Pino
Browse files

Satisfy linters (black, flake, isort)

parent 61ff514c

src/edgecloud/clients/aeros/client.py

+12 −2
Original line number Diff line number Diff line 
# -*- coding: utf-8 -*-

# Mocked API for testing purposes

from typing import Dict, List, Optional



from src.edgecloud.core.edgecloud_interface import EdgeCloudManagementInterface





class EdgeApplicationManager(EdgeCloudManagementInterface):

    def __init__(self, base_url: str):

        self.base_url = base_url
@@ -22,11 +25,18 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
    def deploy_app(self, app_id: str, app_zones: List[Dict]) -> Dict:

        return {"appInstanceId": "abcd-efgh"}



    def get_all_deployed_apps(self, app_id: Optional[str] = None, app_instance_id: Optional[str] = None, region: Optional[str] = None) -> List[Dict]:

    def get_all_deployed_apps(

        self,

        app_id: Optional[str] = None,

        app_instance_id: Optional[str] = None,

        region: Optional[str] = None,

    ) -> List[Dict]:

        return [{"appInstanceId": "abcd-efgh", "status": "ready"}]



    def undeploy_app(self, app_instance_id: str) -> None:

        print(f"Deleting app instance: {app_instance_id}")



    def get_edge_cloud_zones(self, region: Optional[str] = None, status: Optional[str] = None) -> List[Dict]:

    def get_edge_cloud_zones(

        self, region: Optional[str] = None, status: Optional[str] = None

    ) -> List[Dict]:

        return [{"edgeCloudZoneId": "zone-1", "status": "active"}]

src/edgecloud/clients/errors.py

+2 −0
Original line number Diff line number Diff line 
# -*- coding: utf-8 -*-





class EdgeCloudPlatformError(Exception):

    pass

src/edgecloud/clients/i2edge/client.py

+6 −1
Original line number Diff line number Diff line
@@ -142,7 +142,12 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
    def deploy_app(self, app_id: str, app_zones: List[Dict]) -> Dict:

        return {"appInstanceId": "abcd-efgh"}



    def get_all_deployed_apps(self, app_id: Optional[str] = None, app_instance_id: Optional[str] = None, region: Optional[str] = None) -> List[Dict]:

    def get_all_deployed_apps(

        self,

        app_id: Optional[str] = None,

        app_instance_id: Optional[str] = None,

        region: Optional[str] = None,

    ) -> List[Dict]:

        return [{"appInstanceId": "abcd-efgh", "status": "ready"}]



    def undeploy_app(self, app_instance_id: str) -> None:

src/edgecloud/clients/i2edge/common.py

+5 −13
Original line number Diff line number Diff line
@@ -13,10 +13,10 @@ import json 
from typing import Optional



import requests

from src.edgecloud.clients.errors import EdgeCloudPlatformError

from pydantic import BaseModel



from src import logger

from src.edgecloud.clients.errors import EdgeCloudPlatformError



log = logger.get_logger(__name__)
@@ -51,9 +51,7 @@ def i2edge_post(url: str, model_payload: BaseModel) -> dict: 
        return response.json()

    except requests.exceptions.HTTPError as e:

        i2edge_err_msg = get_error_message_from(response)

        err_msg = "Failed to deploy app: {}. Detail: {}".format(

            i2edge_err_msg, e

        )

        err_msg = "Failed to deploy app: {}. Detail: {}".format(i2edge_err_msg, e)

        log.error(err_msg)

        raise I2EdgeError(err_msg)
@@ -70,9 +68,7 @@ def i2edge_post_multiform_data(url: str, model_payload: BaseModel) -> dict: 
        return response.json()

    except requests.exceptions.HTTPError as e:

        i2edge_err_msg = get_error_message_from(response)

        err_msg = "Failed to deploy app: {}. Detail: {}".format(

            i2edge_err_msg, e

        )

        err_msg = "Failed to deploy app: {}. Detail: {}".format(i2edge_err_msg, e)

        log.error(err_msg)

        raise I2EdgeError(err_msg)
@@ -86,9 +82,7 @@ def i2edge_delete(url: str, id: str) -> dict: 
        return response.json()

    except requests.exceptions.HTTPError as e:

        i2edge_err_msg = get_error_message_from(response)

        err_msg = "Failed to undeploy app: {}. Detail: {}".format(

            i2edge_err_msg, e

        )

        err_msg = "Failed to undeploy app: {}. Detail: {}".format(i2edge_err_msg, e)

        log.error(err_msg)

        raise I2EdgeError(err_msg)
@@ -101,8 +95,6 @@ def i2edge_get(url: str, params: Optional[dict]): 
        return response.json()

    except requests.exceptions.HTTPError as e:

        i2edge_err_msg = get_error_message_from(response)

        err_msg = "Failed to get apps: {}. Detail: {}".format(

            i2edge_err_msg, e

        )

        err_msg = "Failed to get apps: {}. Detail: {}".format(i2edge_err_msg, e)

        log.error(err_msg)

        raise I2EdgeError(err_msg)

src/edgecloud/clients/i2edge/schemas.py

+5 −15
Original line number Diff line number Diff line
@@ -121,27 +121,19 @@ class Hugepages(BaseModel): 
class SupportedOSTypes(BaseModel):

    architecture: str = Field(default="x86_64", description="OS architecture")

    distribution: str = Field(default="RHEL", description="OS distribution")

    license: str = Field(

        default="OS_LICENSE_TYPE_FREE", description="OS license type"

    )

    version: str = Field(

        default="OS_VERSION_UBUNTU_2204_LTS", description="OS version"

    )

    license: str = Field(default="OS_LICENSE_TYPE_FREE", description="OS license type")

    version: str = Field(default="OS_VERSION_UBUNTU_2204_LTS", description="OS version")





class FlavourSupported(BaseModel):

    cpuArchType: str = Field(

        default="ISA_X86", description="CPU architecture type"

    )

    cpuArchType: str = Field(default="ISA_X86", description="CPU architecture type")

    cpuExclusivity: bool = Field(default=True, description="CPU exclusivity")

    fpga: int = Field(default=0, description="Number of FPGAs")

    gpu: Optional[List[GPU]] = Field(default=None, description="List of GPUs")

    hugepages: List[Hugepages] = Field(

        default_factory=lambda: [Hugepages()], description="List of hugepages"

    )

    memorySize: str = Field(

        ..., description="Memory size (e.g., '1024MB' or '2GB')"

    )

    memorySize: str = Field(..., description="Memory size (e.g., '1024MB' or '2GB')")

    numCPU: int = Field(..., description="Number of CPUs")

    storageSize: int = Field(default=0, description="Storage size in GB")

    supportedOSTypes: List[SupportedOSTypes] = Field(
@@ -158,9 +150,7 @@ class FlavourSupported(BaseModel): 
        try:

            int(v[:-2])

        except ValueError:

            raise ValueError(

                "memorySize must be a number followed by MB or GB"

            )

            raise ValueError("memorySize must be a number followed by MB or GB")

        return v