Fix format - satisfy linters (black, isort, flake8) (1b0ad0b0) · Commits · OOP / code / transformation-function-sdk

src/edgecloud/clients/i2edge/common.py

+16 −5

Original line number	Diff line number	Diff line
		@@ -39,7 +39,10 @@ def get_error_message_from(response: requests.Response) -> str:


		def i2edge_post(url: str, model_payload: BaseModel) -> dict:
		headers = {"Content-Type": "application/json", "accept": "application/json"}
		headers = {
		"Content-Type": "application/json",
		"accept": "application/json",
		}
		json_payload = json.dumps(model_payload.model_dump(mode="json"))
		try:
		response = requests.post(url, data=json_payload, headers=headers)
		@@ -47,7 +50,9 @@ 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)

		@@ -64,7 +69,9 @@ 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)

		@@ -78,7 +85,9 @@ 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)

		@@ -91,6 +100,8 @@ 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

+15 −5

Original line number	Diff line number	Diff line
		@@ -121,19 +121,27 @@ 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(
		@@ -150,7 +158,9 @@ 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

src/edgecloud/clients/i2edge/utils.py

+9 −3

Original line number	Diff line number	Diff line
		@@ -63,7 +63,9 @@ def instantiate_app_with(
		)
		return flavour_id, application_k8s_namespace
		except I2EdgeError as e:
		err_msg = "Error instantiating app {} in zone {}".format(camara_app_id, zone_id)
		err_msg = "Error instantiating app {} in zone {}".format(
		camara_app_id, zone_id
		)
		log.error("{}. Detailed error: {}".format(err_msg, e))
		raise e

		@@ -93,7 +95,9 @@ def onboard_app_with(
		token=token,
		)

		i2edge.onboard_app(app_id=str(application_id), artefact_id=str(application_id))
		i2edge.onboard_app(
		app_id=str(application_id), artefact_id=str(application_id)
		)
		except I2EdgeError as e:
		err_msg = "Error onboarding app {} in i2edge".format(app_name)
		log.error("{}. Detailed error: {}".format(err_msg, e))
		@@ -112,7 +116,9 @@ def delete_app_instance_by(
		i2edge.delete_flavour(flavour_id=str(flavour_id), zone_id=zone_id)


		def get_app_name_from(namespace: str, i2edge: I2EdgeClient) -> Union[str, None]:
		def get_app_name_from(
		namespace: str, i2edge: I2EdgeClient
		) -> Union[str, None]:
		try:
		response = i2edge.get_all_deployed_apps()
		for deployment in response:

src/edgecloud/clients/piedge/client.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@
		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

src/edgecloud/core/edgecloud_factory.py

+3 −1

Original line number	Diff line number	Diff line
		@@ -29,7 +29,9 @@ class EdgeCloudFactory:
		"""

		@staticmethod
		def create_edgecloud_client(client_name: str, base_url: str) -> EdgeCloudInterface:
		def create_edgecloud_client(
		client_name: str, base_url: str
		) -> EdgeCloudInterface:
		try:
		return EdgeCloudTypes.edgecloud_types[client_name](base_url)
		except KeyError: