Satisfy linters (58d73b9a) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/common/adapters_factory.py

+2 −6

Original line number	Diff line number	Diff line
		@@ -50,12 +50,8 @@ def _network_adapters_factory(client_name: str, base_url: str, **kwargs):
		scs_as_id = kwargs.pop("scs_as_id")

		network_factory = {
		"open5gs": lambda url, scs_id, **kw: Open5GSClient(
		base_url=url, scs_as_id=scs_id, **kw
		),
		"oai": lambda url, scs_id, **kw: OaiCoreClient(
		base_url=url, scs_as_id=scs_id, **kw
		),
		"open5gs": lambda url, scs_id, kw: Open5GSClient(base_url=url, scs_as_id=scs_id, kw),
		"oai": lambda url, scs_id, kw: OaiCoreClient(base_url=url, scs_as_id=scs_id, kw),
		"open5gcore": lambda url, scs_id, **kw: Open5GCoreClient(
		base_url=url, scs_as_id=scs_id, **kw
		),

src/sunrise6g_opensdk/common/sdk.py

+1 −3

Original line number	Diff line number	Diff line
		@@ -56,9 +56,7 @@ class Sdk:
		base_url = config["base_url"]

		# Support of additional paramaters for specific adapters
		kwargs = {
		k: v for k, v in config.items() if k not in ("client_name", "base_url")
		}
		kwargs = {k: v for k, v in config.items() if k not in ("client_name", "base_url")}

		client = sdk_client.instantiate_and_retrieve_adapters(
		domain, client_name, base_url, **kwargs

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

+14 −42

Original line number	Diff line number	Diff line
		@@ -53,9 +53,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		raise EdgeCloudPlatformError("Missing 'appId' in app manifest")

		if app_id in self._app_store:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' already exists"
		)
		raise EdgeCloudPlatformError(f"Application with id '{app_id}' already exists")

		self._app_store[app_id] = app_manifest
		self.logger.debug("Onboarded application with id: %s", app_id)
		@@ -67,17 +65,13 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		def get_onboarded_app(self, app_id: str) -> Dict:
		if app_id not in self._app_store:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' does not exist"
		)
		raise EdgeCloudPlatformError(f"Application with id '{app_id}' does not exist")
		self.logger.debug("Retrieved application with id: %s", app_id)
		return self._app_store[app_id]

		def delete_onboarded_app(self, app_id: str) -> None:
		if app_id not in self._app_store:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' does not exist"
		)
		raise EdgeCloudPlatformError(f"Application with id '{app_id}' does not exist")
		service_instances = self._stopped_services.get(app_id, [])
		self.logger.debug(
		"Deleting application with id: %s and instances: %s",
		@@ -86,29 +80,21 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		)
		for service_instance in service_instances:
		self._purge_deployed_app_from_continuum(service_instance)
		self.logger.debug(
		"successfully purged service instance: %s", service_instance
		)
		self.logger.debug("successfully purged service instance: %s", service_instance)
		del self._stopped_services[app_id] # Clean up stopped services
		del self._app_store[app_id] # Remove from onboarded apps

		def _generate_service_id(self, app_id: str) -> str:
		return f"urn:ngsi-ld:Service:{app_id}-{uuid.uuid4().hex[:4]}"

		def _generate_tosca_yaml_dict(
		self, app_manifest: Dict, app_zones: List[Dict]
		) -> Dict:
		def _generate_tosca_yaml_dict(self, app_manifest: Dict, app_zones: List[Dict]) -> Dict:
		component = app_manifest.get("componentSpec", [{}])[0]
		component_name = component.get("componentName", "application")

		image_path = app_manifest.get("appRepo", {}).get("imagePath", "")
		image_file = image_path.split("/")[-1]
		repository_url = (
		"/".join(image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"
		)
		zone_id = (
		app_zones[0].get("EdgeCloudZone", {}).get("edgeCloudZoneId", "default-zone")
		)
		repository_url = "/".join(image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"
		zone_id = app_zones[0].get("EdgeCloudZone", {}).get("edgeCloudZoneId", "default-zone")

		# Extract minNodeMemory
		min_node_memory = (
		@@ -124,9 +110,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		interface_id = iface.get("interfaceId", "default")
		protocol = iface.get("protocol", "TCP").lower()
		port = iface.get("port", 8080)
		ports[interface_id] = {
		"properties": {"protocol": [protocol], "source": port}
		}
		ports[interface_id] = {"properties": {"protocol": [protocol], "source": port}}

		expose_ports = any(
		iface.get("visibilityType") == "VISIBILITY_EXTERNAL"
		@@ -159,13 +143,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"properties": {
		"cpu_arch": {"equal": "x64"},
		"realtime": {"equal": False},
		"cpu_usage": {
		"less_or_equal": "0.1"
		},
		"cpu_usage": {"less_or_equal": "0.1"},
		"mem_size": {
		"greater_or_equal": str(
		min_node_memory
		)
		"greater_or_equal": str(min_node_memory)
		},
		"domain_id": {"equal": zone_id},
		}
		@@ -203,9 +183,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		# 1. Get app CAMARA manifest
		app_manifest = self._app_store.get(app_id)
		if not app_manifest:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' does not exist"
		)
		raise EdgeCloudPlatformError(f"Application with id '{app_id}' does not exist")

		# 2. Generate unique service ID
		service_id = self._generate_service_id(app_id)
		@@ -306,9 +284,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		for domain in aeros_domains
		]

		def get_edge_cloud_zones_details(
		self, zone_id: str, flavour_id: Optional[str] = None
		) -> Dict:
		def get_edge_cloud_zones_details(self, zone_id: str, flavour_id: Optional[str] = None) -> Dict:
		"""
		Get details of a specific edge cloud zone.
		:param zone_id: The ID of the edge cloud zone
		@@ -350,9 +326,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		# #
		# }
		aeros_client = ContinuumClient(self.base_url)
		ngsild_params = (
		f'format=simplified&type=InfrastructureElement&q=domain=="{zone_id}"'
		)
		ngsild_params = f'format=simplified&type=InfrastructureElement&q=domain=="{zone_id}"'
		self.logger.debug(
		"Querying infrastructure elements for zone %s with params: %s",
		zone_id,
		@@ -548,9 +522,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"""
		pass

		def get_deployed_app_gsma(
		self, app_id: str, app_instance_id: str, zone_id: str
		) -> Dict:
		def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Dict:
		"""
		Retrieves an application instance details from partner OP.

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

+1 −3

Original line number	Diff line number	Diff line
		@@ -151,9 +151,7 @@ class ContinuumClient:
		onboard_url = f"{self.api_url}/hlo_fe/services/{service_id}"
		if config.DEBUG:
		self.logger.debug("Onboard service URL: %s", onboard_url)
		self.logger.debug(
		"Onboard service request body (TOSCA-YAML): %s", tosca_str
		)
		self.logger.debug("Onboard service request body (TOSCA-YAML): %s", tosca_str)
		response = requests.post(
		onboard_url, data=tosca_str, headers=self.hlo_onboard_headers, timeout=15
		)

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

+17 −51

Original line number	Diff line number	Diff line
		@@ -18,9 +18,7 @@ from sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas
		from sunrise6g_opensdk.edgecloud.core.edgecloud_interface import (
		EdgeCloudManagementInterface,
		)
		from sunrise6g_opensdk.edgecloud.core.utils import (
		build_custom_http_response,
		)
		from sunrise6g_opensdk.edgecloud.core.utils import build_custom_http_response

		from ...adapters.i2edge import schemas as i2edge_schemas
		from .common import (
		@@ -78,13 +76,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		zone = camara_schemas.EdgeCloudZone(
		# edgeCloudZoneId = camara_schemas.EdgeCloudZoneId(z["zoneId"]),
		edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(z["zoneId"]),
		edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(
		z["nodeName"]
		),
		edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(z["nodeName"]),
		edgeCloudProvider=camara_schemas.EdgeCloudProvider("i2edge"),
		edgeCloudRegion=camara_schemas.EdgeCloudRegion(
		z["geographyDetails"]
		),
		edgeCloudRegion=camara_schemas.EdgeCloudRegion(z["geographyDetails"]),
		edgeCloudZoneStatus=camara_schemas.EdgeCloudZoneStatus.unknown,
		)
		camara_response.append(zone)
		@@ -258,9 +252,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		i2edge_response.raise_for_status()

		# 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))

		log.info("App onboarded successfully")
		return build_custom_http_response(
		@@ -449,9 +441,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		appId=appId,
		appProviderId=appProviderId,
		appVersion=appVersion,
		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)
		payload = i2edge_schemas.AppDeploy(
		@@ -478,9 +468,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		)

		# CAMARA spec requires appInstances array wrapper
		camara_response = {
		"appInstances": [app_instance_info.model_dump(mode="json")]
		}
		camara_response = {"appInstances": [app_instance_info.model_dump(mode="json")]}

		log.info("App deployment request submitted successfully")
		return build_custom_http_response(
		@@ -642,17 +630,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		for item in response_json:
		content = {
		"zoneId": item.get("zoneId"),
		"reservedComputeResources": item.get(
		"reservedComputeResources"
		),
		"computeResourceQuotaLimits": item.get(
		"computeResourceQuotaLimits"
		),
		"reservedComputeResources": item.get("reservedComputeResources"),
		"computeResourceQuotaLimits": item.get("computeResourceQuotaLimits"),
		"flavoursSupported": item.get("flavoursSupported"),
		"networkResources": item.get("networkResources"),
		"zoneServiceLevelObjsInfo": item.get(
		"zoneServiceLevelObjsInfo"
		),
		"zoneServiceLevelObjsInfo": item.get("zoneServiceLevelObjsInfo"),
		}
		response_list.append(content)
		return build_custom_http_response(
		@@ -683,17 +665,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		response_json = response.json()
		content = {
		"zoneId": response_json.get("zoneId"),
		"reservedComputeResources": response_json.get(
		"reservedComputeResources"
		),
		"computeResourceQuotaLimits": response_json.get(
		"computeResourceQuotaLimits"
		),
		"reservedComputeResources": response_json.get("reservedComputeResources"),
		"computeResourceQuotaLimits": response_json.get("computeResourceQuotaLimits"),
		"flavoursSupported": response_json.get("flavoursSupported"),
		"networkResources": response_json.get("networkResources"),
		"zoneServiceLevelObjsInfo": response_json.get(
		"zoneServiceLevelObjsInfo"
		),
		"zoneServiceLevelObjsInfo": response_json.get("zoneServiceLevelObjsInfo"),
		}
		return build_custom_http_response(
		status_code=200,
		@@ -894,9 +870,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"""
		pass

		def delete_onboarded_app_gsma(
		self, federation_context_id: str, app_id: str
		) -> Response:
		def delete_onboarded_app_gsma(self, federation_context_id: str, app_id: str) -> Response:
		"""
		Deboards an application from specific partner OP zones using GSMA federation.

		@@ -965,9 +939,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		except KeyError as e:
		raise I2EdgeError(f"Missing required field in GSMA deployment payload: {e}")

		def get_deployed_app_gsma(
		self, app_id: str, app_instance_id: str, zone_id: str
		) -> Response:
		def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Response:
		"""
		Retrieves an application instance details from partner OP using GSMA federation.

		@@ -977,9 +949,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:return: Response with application instance details.
		"""
		try:
		url = "{}/application_instance/{}/{}".format(
		self.base_url, zone_id, app_instance_id
		)
		url = "{}/application_instance/{}/{}".format(self.base_url, zone_id, app_instance_id)
		params = {}
		response = i2edge_get(url, params=params)
		if response.status_code == 200:
		@@ -1042,9 +1012,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		except KeyError as e:
		raise I2EdgeError(f"Error retrieving apps: {e}")

		def undeploy_app_gsma(
		self, app_id: str, app_instance_id: str, zone_id: str
		) -> Response:
		def undeploy_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Response:
		"""
		Terminate an application instance on a partner OP zone.

		@@ -1059,9 +1027,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		if response.status_code == 200:
		return build_custom_http_response(
		status_code=200,
		content={
		"response": "Application instance termination request accepted"
		},
		content={"response": "Application instance termination request accepted"},
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,
		url=response.url,