Merge pull request #153 from SunriseOpenOperatorPlatform/feature/make-i2edge-fully-camara-compliant (46a0ca2d) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/edgecloud/adapters/aeros/client.py

+16 −0

Original line number	Original line	Diff line number	Diff line
	@@ -9,6 +9,7 @@ import uuid
	from typing import Any, Dict, List, Optional		from typing import Any, Dict, List, Optional

	import yaml		import yaml
			from requests import Response

	from sunrise6g_opensdk.edgecloud.adapters.aeros import config		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
	from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_client import ContinuumClient		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_client import ContinuumClient
	@@ -223,6 +224,21 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	deployed.append({"appId": stored_app_id, "appInstanceId": instance_id})		deployed.append({"appId": stored_app_id, "appInstanceId": instance_id})
	return deployed		return deployed

			def get_deployed_app(
			self, app_instance_id: str, app_id: Optional[str] = None, region: Optional[str] = None
			) -> Response:
			"""
			Placeholder implementation for CAMARA compliance.
			Retrieves information of a specific application instance.

			:param app_instance_id: Unique identifier of the application instance
			:param app_id: Optional filter by application ID
			:param region: Optional filter by Edge Cloud region
			:return: Response with application instance details
			"""
			# TODO: Implement actual aeros-specific logic for retrieving a specific deployed app
			raise NotImplementedError("get_deployed_app is not yet implemented for aeros adapter")

	def _purge_deployed_app_from_continuum(self, app_id: str) -> None:		def _purge_deployed_app_from_continuum(self, app_id: str) -> None:
	aeros_client = ContinuumClient(self.base_url)		aeros_client = ContinuumClient(self.base_url)
	response = aeros_client.purge_service(app_id)		response = aeros_client.purge_service(app_id)

src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py

+271 −141

Original line number	Original line	Diff line number	Diff line
	@@ -7,6 +7,7 @@
	# - Sergio Giménez (sergio.gimenez@i2cat.net)		# - Sergio Giménez (sergio.gimenez@i2cat.net)
	# - César Cajas (cesar.cajas@i2cat.net)		# - César Cajas (cesar.cajas@i2cat.net)
	##		##
			import json
	from copy import deepcopy		from copy import deepcopy
	from typing import Dict, List, Optional		from typing import Dict, List, Optional

	@@ -84,9 +85,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	params = {}		params = {}

	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params) # expects 200 by default
	if response.status_code == 200:
	response.raise_for_status()
	i2edge_response = response.json()		i2edge_response = response.json()
	log.info("Availability zones retrieved successfully")		log.info("Availability zones retrieved successfully")
	# Normalise to CAMARA format		# Normalise to CAMARA format
	@@ -100,7 +99,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response
	except KeyError as e:		except KeyError as e:
	log.error(f"Missing required CAMARA field in app manifest: {e}")		log.error(f"Missing required CAMARA field in app manifest: {e}")
	raise ValueError(f"Invalid CAMARA manifest – missing field: {e}")		raise ValueError(f"Invalid CAMARA manifest – missing field: {e}")
	@@ -111,6 +109,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	# ------------------------------------------------------------------------		# ------------------------------------------------------------------------
	# Artefact Management (i2Edge-Specific, Non-CAMARA)		# Artefact Management (i2Edge-Specific, Non-CAMARA)
	# ------------------------------------------------------------------------		# ------------------------------------------------------------------------
			# All artefact methods now return Response objects for API consistency

	def create_artefact(		def create_artefact(
	self,		self,
	@@ -122,7 +121,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	password: Optional[str] = None,		password: Optional[str] = None,
	token: Optional[str] = None,		token: Optional[str] = None,
	user_name: Optional[str] = None,		user_name: Optional[str] = None,
	):		) -> Response:
	"""		"""
	Creates an artefact in the i2Edge platform.		Creates an artefact in the i2Edge platform.
	This is an i2Edge-specific operation not covered by CAMARA standards.		This is an i2Edge-specific operation not covered by CAMARA standards.
	@@ -159,13 +158,13 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		raise e

	def get_artefact(self, artefact_id: str) -> Dict:		def get_artefact(self, artefact_id: str) -> Response:
	"""		"""
	Retrieves details about a specific artefact.		Retrieves details about a specific artefact.
	This is an i2Edge-specific operation not covered by CAMARA standards.		This is an i2Edge-specific operation not covered by CAMARA standards.

	:param artefact_id: Unique identifier of the artefact		:param artefact_id: Unique identifier of the artefact
	:return: Dictionary with artefact details		:return: Response with artefact details
	"""		"""
	url = "{}/artefact/{}".format(self.base_url, artefact_id)		url = "{}/artefact/{}".format(self.base_url, artefact_id)
	params = {}		params = {}
	@@ -176,12 +175,12 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		raise e

	def get_all_artefacts(self) -> List[Dict]:		def get_all_artefacts(self) -> Response:
	"""		"""
	Retrieves a list of all artefacts.		Retrieves a list of all artefacts.
	This is an i2Edge-specific operation not covered by CAMARA standards.		This is an i2Edge-specific operation not covered by CAMARA standards.

	:return: List of artefact details		:return: Response with list of artefact details
	"""		"""
	url = "{}/artefact".format(self.base_url)		url = "{}/artefact".format(self.base_url)
	params = {}		params = {}
	@@ -192,7 +191,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		raise e

	def delete_artefact(self, artefact_id: str):		def delete_artefact(self, artefact_id: str) -> Response:
	"""		"""
	Deletes a specific artefact from the i2Edge platform.		Deletes a specific artefact from the i2Edge platform.
	This is an i2Edge-specific operation not covered by CAMARA standards.		This is an i2Edge-specific operation not covered by CAMARA standards.
	@@ -255,11 +254,8 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	i2edge_response = i2edge_post(		i2edge_response = i2edge_post(
	f"{self.base_url}/application/onboarding",		f"{self.base_url}/application/onboarding",
	model_payload=i2edge_payload,		model_payload=i2edge_payload,
			expected_status=201,
	)		)
	# OpenAPI specifies 201 for successful application onboarding
	if i2edge_response.status_code == 201:
	i2edge_response.raise_for_status()

	# Build CAMARA-compliant response using schema		# Build CAMARA-compliant response using schema
	submitted_app = camara_schemas.SubmittedApp(appId=camara_schemas.AppId(app_id))		submitted_app = camara_schemas.SubmittedApp(appId=camara_schemas.AppId(app_id))

	@@ -272,9 +268,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=i2edge_response.url,		url=i2edge_response.url,
	request=i2edge_response.request,		request=i2edge_response.request,
	)		)
	else:
	i2edge_response.raise_for_status()
	# TODO: Implement CAMARA-compliant error handling for failed onboarding responses
	except ValidationError as e:		except ValidationError as e:
	error_details = "; ".join(		error_details = "; ".join(
	[f"Field '{err['loc'][0]}': {err['msg']}" for err in e.errors()]		[f"Field '{err['loc'][0]}': {err['msg']}" for err in e.errors()]
	@@ -295,9 +288,8 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	"""		"""
	url = "{}/application/onboarding".format(self.base_url)		url = "{}/application/onboarding".format(self.base_url)
	try:		try:
	response = i2edge_delete(url, app_id)		# i2Edge returns 200 for successful deletions, but CAMARA expects 204
	response.raise_for_status()		response = i2edge_delete(url, app_id, expected_status=200)

	log.info("App onboarded deleted successfully")		log.info("App onboarded deleted successfully")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=204,		status_code=204,
	@@ -322,8 +314,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url = "{}/application/onboarding/{}".format(self.base_url, app_id)		url = "{}/application/onboarding/{}".format(self.base_url, app_id)
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params) # expects 200 by default
	response.raise_for_status()
	i2edge_response = response.json()		i2edge_response = response.json()

	# Extract and transform i2Edge response to CAMARA format		# Extract and transform i2Edge response to CAMARA format
	@@ -373,8 +364,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url = "{}/applications/onboarding".format(self.base_url)		url = "{}/applications/onboarding".format(self.base_url)
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params) # expects 200 by default
	response.raise_for_status()
	i2edge_response = response.json()		i2edge_response = response.json()

	# Transform i2Edge response to CAMARA format using AppManifest schema		# Transform i2Edge response to CAMARA format using AppManifest schema
	@@ -390,6 +380,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	name=app_metadata.get("appName", ""),		name=app_metadata.get("appName", ""),
	version=app_metadata.get("version", ""),		version=app_metadata.get("version", ""),
	appProvider=profile_data.get("appProviderId", ""),		appProvider=profile_data.get("appProviderId", ""),
			# Hardcoding mandatory fields that don't exist in i2Edge
	packageType="CONTAINER",		packageType="CONTAINER",
	appRepo={"type": "PUBLICREPO", "imagePath": "not-available"},		appRepo={"type": "PUBLICREPO", "imagePath": "not-available"},
	requiredResources={		requiredResources={
	@@ -456,15 +447,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	zoneInfo=i2edge_schemas.ZoneInfoRef(flavourId=self.flavour_id, zoneId=zone_id),		zoneInfo=i2edge_schemas.ZoneInfoRef(flavourId=self.flavour_id, zoneId=zone_id),
	)		)
	url = "{}/application_instance".format(self.base_url)		url = "{}/application_instance".format(self.base_url)
	payload = i2edge_schemas.AppDeploy(		payload = i2edge_schemas.AppDeploy(app_deploy_data=app_deploy_data)
	app_deploy_data=app_deploy_data, app_parameters={"namespace": "test"}
	)

	# Deployment request to i2Edge		# Deployment request to i2Edge - CAMARA expects 202 for deployment
	try:		try:
	i2edge_response = i2edge_post(url, payload)		i2edge_response = i2edge_post(url, payload, expected_status=202)
	if i2edge_response.status_code == 202:
	i2edge_response.raise_for_status()
	i2edge_data = i2edge_response.json()		i2edge_data = i2edge_response.json()

	# Build CAMARA-compliant response		# Build CAMARA-compliant response
	@@ -491,70 +478,216 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=i2edge_response.url,		url=i2edge_response.url,
	request=i2edge_response.request,		request=i2edge_response.request,
	)		)
	else:
	i2edge_response.raise_for_status()
	except I2EdgeError as e:		except I2EdgeError as e:
	log.error(f"Failed to deploy app to i2Edge: {e}")		log.error(f"Failed to deploy app to i2Edge: {e}")
	raise		raise

	def get_all_deployed_apps(self) -> List[Dict]:		def get_all_deployed_apps(
			self,
			app_id: Optional[str] = None,
			app_instance_id: Optional[str] = None,
			region: Optional[str] = None,
			) -> Response:
	"""		"""
	Retrieves information of all application instances.		Retrieves information of all application instances using CAMARA-compliant interface.
			Returns a CAMARA-compliant response.

	:return: List of application instance details		:param app_id: Filter by application ID
			:param app_instance_id: Filter by instance ID
			:param region: Filter by Edge Cloud region
			:return: Response with application instance details in CAMARA format
	"""		"""
	url = "{}/application_instances".format(self.base_url)		url = "{}/application_instances".format(self.base_url)
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:		i2edge_response = response.json()
	response.raise_for_status()
			# Transform i2Edge response to CAMARA format
			camara_instances = []
			if isinstance(i2edge_response, list):
			for instance_data in i2edge_response:
			# Apply filters if provided
			if app_id and instance_data.get("app_id") != app_id:
			continue
			if app_instance_id and instance_data.get("app_instance_id") != app_instance_id:
			continue
			if region and instance_data.get("region") != region:
			continue

			# Transform to CAMARA AppInstanceInfo
			try:
			# Map i2Edge status to CAMARA status
			i2edge_status = instance_data.get("deploy_status", "unknown")
			camara_status = "ready" if i2edge_status == "DEPLOYED" else "unknown"

			# Extract zone_id from app_spec.nodeSelector
			zone_id = "unknown"
			app_spec = instance_data.get("app_spec", {})
			node_selector = app_spec.get("nodeSelector", {})
			if "feature.node.kubernetes.io/zoneID" in node_selector:
			zone_id = node_selector["feature.node.kubernetes.io/zoneID"]

			app_instance_info = camara_schemas.AppInstanceInfo(
			name=camara_schemas.AppInstanceName(
			instance_data.get("app_instance_id", "unknown")
			),
			appId=camara_schemas.AppId(instance_data.get("app_id", "unknown")),
			appInstanceId=camara_schemas.AppInstanceId(
			instance_data.get("app_instance_id", "unknown")
			),
			appProvider=camara_schemas.AppProvider(
			instance_data.get("app_provider", "Unknown_Provider")
			),
			status=camara_schemas.Status(
			camara_status
			), # Map the i2Edge "DEPLOYED" status to the CAMARA "ready" status for consistency with CAMARA specifications.
			edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(
			zone_id
			), # FIX: Extract from nodeSelector
			)
			camara_instances.append(app_instance_info.model_dump(mode="json"))
			except Exception as validation_error:
			# Skip instances that fail validation
			log.warning(f"Skipping invalid instance data: {validation_error}")
			continue

			# CAMARA spec format for multiple instances response
			camara_response = {"appInstances": camara_instances}

	log.info("All app instances retrieved successfully")		log.info("All app instances retrieved successfully")
	return response.json()		return build_custom_http_response(
	return response		status_code=response.status_code,
			content=camara_response,
			headers={"Content-Type": "application/json"},
			encoding="utf-8",
			url=response.url,
			request=response.request,
			)
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		log.error(f"Failed to retrieve all app instances from i2Edge: {e}")
			raise

	def get_deployed_app(self, app_id, zone_id) -> List[Dict]:		def get_deployed_app(
			self, app_instance_id: str, app_id: Optional[str] = None, region: Optional[str] = None
			) -> Response:
	"""		"""
	Retrieves a specific deployed application instance by app ID and zone ID.		Retrieves information of a specific application instance using CAMARA-compliant interface.
			Returns a CAMARA-compliant response.

	:param app_id: Unique identifier of the application		:param app_instance_id: Unique identifier of the application instance (mandatory)
	:param zone_id: Unique identifier of the Edge Cloud Zone		:param app_id: Optional filter by application ID for validation
	:return: Application instance details or None if not found		:param region: Optional filter by Edge Cloud region for validation
			:return: Response with application instance details in CAMARA format
	"""		"""
	# Logic: Get all onboarded apps and filter the one where release_name == artifact name

	# Step 1) Extract "app_name" from the onboarded app using the "app_id"
	try:		try:
	onboarded_app_response = self.get_onboarded_app(app_id)		# Get raw i2Edge data without CAMARA filtering to find the zone_id
	onboarded_app_response.raise_for_status()		url = "{}/application_instances".format(self.base_url)
	onboarded_app_data = onboarded_app_response.json()		raw_response = i2edge_get(url, params={}, expected_status=200)
	except I2EdgeError as e:		raw_instances = raw_response.json()
	log.error(f"Failed to retrieve app data: {e}")
	raise ValueError(f"No onboarded app found with ID: {app_id}")		# Find the specific instance in raw data to get its zone_id
			target_zone_id = None
			original_instance = None
			if isinstance(raw_instances, list):
			for instance_data in raw_instances:
			if instance_data.get("app_instance_id") == app_instance_id:
			# Optional validation: check app_id if provided
			if app_id and instance_data.get("app_id") != app_id:
			log.warning(
			f"App instance {app_instance_id} found but app_id mismatch: expected {app_id}, found {instance_data.get('app_id')}"
			)
			continue

			# Optional validation: check region if provided
			if region and instance_data.get("region") != region:
			log.warning(
			f"App instance {app_instance_id} found but region mismatch: expected {region}, found {instance_data.get('region')}"
			)
			continue

			target_zone_id = instance_data.get("zone_id")
			original_instance = instance_data
			break

			# If instance not found in list, try to get a fallback zone dynamically
			if not target_zone_id:
			log.warning(
			f"App instance {app_instance_id} not found in instances list, attempting to find fallback zone"
			)

			# Try to get available zones and use the first one as fallback
	try:		try:
	# Extract app name from CAMARA response format		zones_response = self.get_edge_cloud_zones()
	app_name = onboarded_app_data.get("name", "")		if zones_response.status_code == 200:
	if not app_name:		zones_data = (
	raise KeyError("name")		zones_response.json()
	except KeyError as e:		if hasattr(zones_response, "json")
	raise ValueError(f"Onboarded app missing required field: {e}")		else json.loads(zones_response.content.decode())
			)
	# Step 2) Retrieve all deployed apps and filter the one(s) where release_name == app_name		if zones_data and len(zones_data) > 0:
	deployed_apps = self.get_all_deployed_apps()		target_zone_id = zones_data[0].get("edgeCloudZoneId")
	if not deployed_apps:		log.info(f"Using fallback zone: {target_zone_id}")
	return []		else:
			raise I2EdgeError("No available zones found for fallback")
	# Filter apps where release_name matches our app_name and zone matches		else:
	for app_instance_name in deployed_apps:		raise I2EdgeError(
	if (		f"Failed to retrieve zones for fallback: {zones_response.status_code}"
	app_instance_name.get("release_name") == app_name		)
	and app_instance_name.get("zone_id") == zone_id		except Exception as zone_error:
	):		log.error(f"Could not retrieve fallback zone: {zone_error}")
	return app_instance_name		raise I2EdgeError(
	return None		f"App instance {app_instance_id} not found and no fallback zone available"
			)

			# Use provided app_id if available, otherwise mark as unknown since we don't have instance data
			fallback_app_id = app_id if app_id else "unknown"
			original_instance = {"app_id": fallback_app_id, "app_provider": "Unknown_Provider"}

			# Now use the correct i2Edge endpoint with zone_id and app_instance_id
			url = f"{self.base_url}/application_instance/{target_zone_id}/{app_instance_id}"
			params = {}
			response = i2edge_get(url, params=params, expected_status=200)
			i2edge_response = response.json()

			# The i2Edge response has different structure: {"accesspointInfo": [...], "appInstanceState": "DEPLOYED"}
			# We need to map this to CAMARA format and get additional info from the raw instance data

			# Transform i2Edge response to CAMARA format
			app_instance_info = camara_schemas.AppInstanceInfo(
			name=camara_schemas.AppInstanceName(app_instance_id),
			appId=camara_schemas.AppId(
			original_instance.get("app_id") if original_instance else "unknown"
			),
			appInstanceId=camara_schemas.AppInstanceId(app_instance_id),
			appProvider=camara_schemas.AppProvider(
			original_instance.get("app_provider", "Unknown_Provider")
			if original_instance
			else "Unknown_Provider"
			),
			status=camara_schemas.Status(
			"ready" if i2edge_response.get("appInstanceState") == "DEPLOYED" else "unknown"
			),
			edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(target_zone_id),
			)

			# CAMARA spec format for single instance response
			camara_response = {"appInstance": app_instance_info.model_dump(mode="json")}

			log.info("App instance retrieved successfully")
			return build_custom_http_response(
			status_code=response.status_code,
			content=camara_response,
			headers={"Content-Type": "application/json"},
			encoding="utf-8",
			url=response.url,
			request=response.request,
			)
			except I2EdgeError as e:
			log.error(
			f"Failed to retrieve app instance from i2Edge (zone_id: {target_zone_id}): {e}"
			)
			raise

	def undeploy_app(self, app_instance_id: str) -> Response:		def undeploy_app(self, app_instance_id: str) -> Response:
	"""		"""
	@@ -566,9 +699,8 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	"""		"""
	url = "{}/application_instance".format(self.base_url)		url = "{}/application_instance".format(self.base_url)
	try:		try:
	i2edge_response = i2edge_delete(url, app_instance_id)		# i2Edge returns 200 for successful deletions, but CAMARA expects 204
	if i2edge_response.status_code == 200:		i2edge_response = i2edge_delete(url, app_instance_id, expected_status=200)
	i2edge_response.raise_for_status()

	log.info("App instance deleted successfully")		log.info("App instance deleted successfully")
	# CAMARA-compliant 204 response (No Content for successful deletion)		# CAMARA-compliant 204 response (No Content for successful deletion)
	@@ -580,8 +712,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=i2edge_response.url,		url=i2edge_response.url,
	request=i2edge_response.request,		request=i2edge_response.request,
	)		)
	else:
	i2edge_response.raise_for_status()
	except I2EdgeError as e:		except I2EdgeError as e:
	log.error(f"Failed to undeploy app from i2Edge: {e}")		log.error(f"Failed to undeploy app from i2Edge: {e}")
	raise		raise

src/sunrise6g_opensdk/edgecloud/adapters/i2edge/common.py

+37 −9

Original line number	Original line	Diff line number	Diff line
	@@ -37,16 +37,28 @@ def get_error_message_from(response: requests.Response) -> str:
	return response.text		return response.text


	def i2edge_post(url: str, model_payload: BaseModel) -> dict:		def i2edge_post(url: str, model_payload: BaseModel, expected_status: int = 201) -> dict:
	headers = {		headers = {
	"Content-Type": "application/json",		"Content-Type": "application/json",
	"accept": "application/json",		"accept": "application/json",
	}		}
	json_payload = json.dumps(model_payload.model_dump(mode="json", exclude_none=True))		json_payload = json.dumps(model_payload.model_dump(mode="json", exclude_none=True))

			# Debug: Log the payload being sent to i2Edge
			log.debug(f"Sending payload to i2Edge: {json_payload}")

	try:		try:
	response = requests.post(url, data=json_payload, headers=headers)		response = requests.post(url, data=json_payload, headers=headers)
	response.raise_for_status()		if response.status_code == expected_status:
	return response		return response
			else:
			# Raise an error with meaningful message about status code mismatch
			i2edge_err_msg = get_error_message_from(response)
			err_msg = "Failed to post: Expected status {}, got {}. Detail: {}".format(
			expected_status, response.status_code, i2edge_err_msg
			)
			log.error(err_msg)
			raise I2EdgeError(err_msg)
	except requests.exceptions.HTTPError as e:		except requests.exceptions.HTTPError as e:
	i2edge_err_msg = get_error_message_from(response)		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)
	@@ -88,13 +100,21 @@ def i2edge_post_multiform_data(url: str, model_payload: BaseModel) -> dict:
	raise I2EdgeError(err_msg)		raise I2EdgeError(err_msg)


	def i2edge_delete(url: str, id: str) -> dict:		def i2edge_delete(url: str, id: str, expected_status: int = 200) -> dict:
	headers = {"accept": "application/json"}		headers = {"accept": "application/json"}
	try:		try:
	query = "{}/{}".format(url, id)		query = "{}/{}".format(url, id)
	response = requests.delete(query, headers=headers)		response = requests.delete(query, headers=headers)
	response.raise_for_status()		if response.status_code == expected_status:
	return response		return response
			else:
			# Raise an error with meaningful message about status code mismatch
			i2edge_err_msg = get_error_message_from(response)
			err_msg = "Failed to delete: Expected status {}, got {}. Detail: {}".format(
			expected_status, response.status_code, i2edge_err_msg
			)
			log.error(err_msg)
			raise I2EdgeError(err_msg)
	except requests.exceptions.HTTPError as e:		except requests.exceptions.HTTPError as e:
	i2edge_err_msg = get_error_message_from(response)		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)
	@@ -102,12 +122,20 @@ def i2edge_delete(url: str, id: str) -> dict:
	raise I2EdgeError(err_msg)		raise I2EdgeError(err_msg)


	def i2edge_get(url: str, params: Optional[dict]):		def i2edge_get(url: str, params: Optional[dict], expected_status: int = 200):
	headers = {"accept": "application/json"}		headers = {"accept": "application/json"}
	try:		try:
	response = requests.get(url, params=params, headers=headers)		response = requests.get(url, params=params, headers=headers)
	response.raise_for_status()		if response.status_code == expected_status:
	return response		return response
			else:
			# Raise an error with meaningful message about status code mismatch
			i2edge_err_msg = get_error_message_from(response)
			err_msg = "Failed to get: Expected status {}, got {}. Detail: {}".format(
			expected_status, response.status_code, i2edge_err_msg
			)
			log.error(err_msg)
			raise I2EdgeError(err_msg)
	except requests.exceptions.HTTPError as e:		except requests.exceptions.HTTPError as e:
	i2edge_err_msg = get_error_message_from(response)		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)