fix(edgecloud): align lite2edge with federated flows (e2291b96) · Commits · OOP / code / transformation-function-sdk

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

+189 −59

Original line number	Diff line number	Diff line
		@@ -21,6 +21,12 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		self.api_url = f"{self.base_url}/api/v1"
		logger.info(f"Initialized Lite2Edge Client with URL: {self.api_url}")

		def _build_json_response(self, status_code: int, payload) -> Response:
		response = Response()
		response.status_code = status_code
		response._content = json.dumps(payload).encode("utf-8")
		return response

		def _convert_gsma_zones_to_camara(self, zones: List[Dict]) -> List[Dict]:
		"""Convert GSMA zone format to CAMARA format"""
		camara_zones = []
		@@ -71,6 +77,8 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"isStandalone": False,
		"version": "1.28",
		}
		if not isinstance(camara_app.get("componentSpec"), list):
		camara_app["componentSpec"] = []
		if "componentSpec" in camara_app:
		for comp in camara_app["componentSpec"]:
		if "networkInterfaces" in comp:
		@@ -168,6 +176,54 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		return response

		def get_edge_cloud_zones_list_gsma(self) -> Response:
		response = self.get_edge_cloud_zones()
		if response.status_code != 200:
		return response

		zones = response.json()
		zone_list = []
		for zone in zones:
		zone_list.append(
		{
		"zoneId": zone.get("edgeCloudZoneId", zone.get("zoneId", "")),
		"edgeCloudRegion": zone.get("edgeCloudRegion", "unknown"),
		}
		)
		return self._build_json_response(200, zone_list)

		def get_edge_cloud_zones_gsma(self) -> Response:
		response = self.get_edge_cloud_zones()
		if response.status_code != 200:
		return response

		zones = response.json()
		detailed_zones = []
		for zone in zones:
		zone_id = zone.get("edgeCloudZoneId", zone.get("zoneId", ""))
		detailed_zones.append(
		{
		"zoneId": zone_id,
		"edgeCloudRegion": zone.get("edgeCloudRegion", "unknown"),
		"reservedComputeResources": [],
		"computeResourceQuotaLimits": [],
		"flavoursSupported": [],
		"networkResources": {},
		"zoneServiceLevelObjsInfo": {},
		}
		)
		return self._build_json_response(200, detailed_zones)

		def get_edge_cloud_zone_details_gsma(self, zone_id: str) -> Response:
		response = self.get_edge_cloud_zones_gsma()
		if response.status_code != 200:
		return response

		for zone in response.json():
		if zone.get("zoneId") == zone_id:
		return self._build_json_response(200, zone)
		return self._build_json_response(404, {"error": "Zone not found"})

		def onboard_app(self, app_manifest: Dict) -> Response:
		return requests.post(f"{self.api_url}/apps/onboard", json=app_manifest)

		@@ -245,13 +301,14 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return requests.delete(f"{self.api_url}/artefacts/{artefact_id}")

		def get_deployed_app(self, app_id: str, app_instance_id: str) -> Response:
		return requests.get(f"{self.api_url}/apps/{app_id}/instances/{app_instance_id}")
		return requests.get(f"{self.api_url}/apps/deployments/{app_instance_id}")

		def get_all_deployed_apps(self) -> Response:
		response = requests.get(f"{self.api_url}/app_instances")
		response = requests.get(f"{self.api_url}/apps/deployments")

		if response.status_code == 200:
		deployed_apps = response.json()
		data = response.json()
		deployed_apps = data.get("appInstances", data) if isinstance(data, dict) else data
		camara_apps = []
		for app_data in deployed_apps:
		if "appInstanceId" in app_data:
		@@ -297,12 +354,20 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return response

		def deploy_app(self, app_id: str, app_deploy_data: Dict) -> Response:
		if isinstance(app_deploy_data, list):
		zones = app_deploy_data
		else:
		zones = app_deploy_data.get("appZones", [])
		if not zones:
		raise ValueError("No appZones provided")

		zone_ref = zones[0].get("EdgeCloudZone", {})
		zone_id = zone_ref.get("edgeCloudZoneId")
		first_zone = zones[0]
		if isinstance(first_zone, dict) and "EdgeCloudZone" in first_zone:
		zone_id = first_zone.get("EdgeCloudZone", {}).get("edgeCloudZoneId")
		elif isinstance(first_zone, dict):
		zone_id = first_zone.get("zoneId") or first_zone.get("edgeCloudZoneId")
		else:
		zone_id = None
		payload = {
		"appId": app_id,
		"zoneId": zone_id,
		@@ -324,7 +389,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return response

		def undeploy_app(self, app_id: str, app_instance_id: str) -> Response:
		response = requests.delete(f"{self.api_url}/apps/{app_id}/instances/{app_instance_id}")
		response = requests.delete(f"{self.api_url}/apps/deployments/{app_instance_id}")

		if response.status_code == 204:
		new_resp = Response()
		@@ -335,13 +400,16 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return response

		def get_app_by_zone(self, app_id: str, zone_id: str) -> Response:
		response = requests.get(f"{self.api_url}/apps/{app_id}/instances", params={"zoneId": zone_id})
		response = self.get_all_deployed_apps()

		if response.status_code == 200:
		instances = response.json()
		data = response.json()
		instances = data.get("appInstances", data) if isinstance(data, dict) else data
		if isinstance(instances, list):
		for instance_data in instances:
		if instance_data.get("zoneId") == zone_id:
		instance_zone_id = instance_data.get("edgeCloudZoneId") or instance_data.get("zoneId")
		if instance_data.get("appId") != app_id or instance_zone_id != zone_id:
		continue
		app_status = instance_data.get("status", "unknown")
		interface_points = []
		for interface in instance_data.get("access_points", []) or []:
		@@ -369,6 +437,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"appId": instance_data.get("appId") or app_id,
		"appInstanceState": app_status,
		"accesspointInfo": interface_points,
		"edgeCloudZoneId": instance_zone_id,
		}
		).encode("utf-8")
		return new_resp
		@@ -377,11 +446,18 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		def get_app_by_zone_app_instance_id(
		self, app_id: str, app_instance_id: str, zone_id: str
		) -> Response:
		response = requests.get(f"{self.api_url}/app_instances/{app_instance_id}")
		response = self.get_deployed_app(app_id, app_instance_id)

		if response.status_code == 200:
		instance_data = response.json()
		app_status = instance_data.get("status", "unknown")
		payload = response.json()
		instance_data = payload.get("appInstance", payload) if isinstance(payload, dict) else payload
		instance_zone_id = instance_data.get("edgeCloudZoneId") or instance_data.get("zoneId", zone_id)
		if instance_data.get("appId") not in (None, app_id) or instance_zone_id != zone_id:
		not_found_resp = Response()
		not_found_resp.status_code = 404
		not_found_resp._content = json.dumps({"error": "App instance not found"}).encode("utf-8")
		return not_found_resp
		app_status = instance_data.get("appInstanceState") or instance_data.get("status", "unknown")
		interface_points = []
		for interface in instance_data.get("access_points", []) or []:
		port = interface.get("port")
		@@ -408,7 +484,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"appId": instance_data.get("appId") or app_id,
		"appInstanceState": app_status,
		"accesspointInfo": interface_points,
		"zoneId": instance_data.get("zoneId", zone_id),
		"edgeCloudZoneId": instance_zone_id,
		}
		).encode("utf-8")
		return new_resp
		@@ -417,22 +493,20 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		def create_artefact_gsma(self, artefact_payload: Dict) -> Response:
		payload = artefact_payload.copy()
		repo_location = payload.get("artefactRepoLocation", {})
		response = self.create_artefact(
		artefact_id=payload.get("artefactId", ""),
		artefact_name=payload.get("artefactName", ""),
		repo_name=repo_location.get("repoURL", ""),
		repo_type=payload.get("repoType", "PUBLICREPO"),
		repo_url=repo_location.get("repoURL", ""),
		user_name=repo_location.get("userName"),
		password=repo_location.get("password"),
		token=repo_location.get("token"),
		)
		response = requests.post(f"{self.api_url}/artefacts", json=payload)

		if response.status_code in {200, 201}:
		response_payload = {}
		try:
		response_payload = response.json()
		except ValueError:
		response_payload = {}

		new_resp = Response()
		new_resp.status_code = 201
		new_resp._content = json.dumps(payload).encode("utf-8")
		merged_payload = response_payload.copy()
		merged_payload.update(payload)
		new_resp._content = json.dumps(merged_payload).encode("utf-8")
		return new_resp
		return response

		@@ -440,9 +514,24 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		response = self.get_artefact(artefact_id)
		if response.status_code == 200:
		data = response.json()
		payload = {
		"artefactId": data.get("artefactId", artefact_id),
		"appProviderId": data.get("appProviderId"),
		"artefactName": data.get("artefactName") or data.get("name"),
		"artefactDescription": data.get("artefactDescription"),
		"artefactVersionInfo": data.get("artefactVersionInfo"),
		"artefactVirtType": data.get("artefactVirtType"),
		"artefactFileName": data.get("artefactFileName"),
		"artefactFileFormat": data.get("artefactFileFormat"),
		"artefactDescriptorType": data.get("artefactDescriptorType"),
		"repoType": data.get("repoType", "PUBLICREPO"),
		"artefactRepoLocation": data.get("artefactRepoLocation")
		or {"repoURL": data.get("repoUrl") or data.get("repoURL") or ""},
		"componentSpec": data.get("componentSpec"),
		}
		new_resp = Response()
		new_resp.status_code = 200
		new_resp._content = json.dumps(data).encode("utf-8")
		new_resp._content = json.dumps(payload).encode("utf-8")
		return new_resp
		return response

		@@ -465,22 +554,63 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return new_resp
		return response

		def get_deployed_app_gsma(self, app_id: str, app_instance_id: str) -> Response:
		def onboard_app_gsma(self, request_body: Dict) -> Response:
		return self.onboard_app(self._convert_gsma_app_manifest_to_camara(request_body))

		def get_onboarded_app_gsma(self, app_id: str) -> Response:
		response = self._get_onboarded_app_raw(app_id)
		if response.status_code != 200:
		return response

		response_data = response.json()
		if "appManifest" in response_data:
		response_data = response_data["appManifest"]
		return self._build_json_response(200, response_data)

		def patch_onboarded_app_gsma(self, app_id: str, request_body: Dict) -> Response:
		return self.update_onboarded_app(app_id, request_body)

		def delete_onboarded_app_gsma(self, app_id: str) -> Response:
		response = self.delete_onboarded_app(app_id)
		if response.status_code in {200, 204}:
		new_resp = Response()
		new_resp.status_code = 204
		new_resp._content = b""
		return new_resp
		return response

		def get_deployed_app_gsma(
		self, app_id: str, app_instance_id: str, zone_id: Optional[str] = None
		) -> Response:
		if zone_id:
		return self.get_app_by_zone_app_instance_id(app_id, app_instance_id, zone_id)
		return self.get_deployed_app(app_id, app_instance_id)

		def deploy_app_gsma(self, app_id: str, app_deploy_data: Dict) -> Response:
		def deploy_app_gsma(self, request_body, app_deploy_data: Optional[Dict] = None) -> Response:
		if app_deploy_data is None:
		app_id = request_body.get("appId", "")
		zone_id = request_body.get("zoneInfo", {}).get("zoneId", "")
		else:
		app_id = request_body
		zone_id = app_deploy_data.get("zoneInfo", {}).get("zoneId", "")

		payload = {
		"appZones": [
		{
		"EdgeCloudZone": {
		"edgeCloudZoneId": app_deploy_data.get("zoneInfo", {}).get("zoneId", "")
		"edgeCloudZoneId": zone_id
		}
		}
		]
		}
		return self.deploy_app(app_id, payload)

		def undeploy_app_gsma(self, app_id: str, app_instance_id: str) -> Response:
		def get_all_deployed_apps_gsma(self) -> Response:
		return self.get_all_deployed_apps()

		def undeploy_app_gsma(
		self, app_id: str, app_instance_id: str, zone_id: Optional[str] = None
		) -> Response:
		response = self.undeploy_app(app_id, app_instance_id)
		if response.status_code == 200:
		new_resp = Response()

Admin message