hotfix/gsma-schemas-and-methods-refactor: use pydantic schemas to check input gsma payloads (2877a8ca) · Commits · OOP / code / transformation-function-sdk

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

+77 −32

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

		from pydantic import ValidationError
		@@ -798,6 +797,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		try:
		response = i2edge_get(url, params=params, expected_status=200)
		response_json = response.json()
		# TODO: fix malformed GPU field in i2Edge, it should be a list of objects, not strings
		# --- Quick fix for malformed GPU entries ---
		quota_limits = response_json.get("computeResourceQuotaLimits", [])
		for item in quota_limits:
		if isinstance(item, dict) and isinstance(item.get("gpu"), list):
		fixed_gpu = []
		for g in item["gpu"]:
		if isinstance(g, str):
		try:
		# Convert single quotes to double quotes for valid JSON
		fixed_gpu.append(json.loads(g.replace("'", '"')))
		except json.JSONDecodeError:
		continue # ignore invalid entries
		else:
		fixed_gpu.append(g)
		item["gpu"] = fixed_gpu
		# --- End quick fix ---
		mapped = map_zone(response_json)
		try:
		validated_data = gsma_schemas.ZoneRegisteredData.model_validate(mapped)
		@@ -829,15 +845,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:return: Response with artefact upload confirmation.
		"""
		try:
		artefact_id = request_body["artefactId"]
		artefact_name = request_body["artefactName"]
		repo_data = request_body["artefactRepoLocation"]
		# Validate input body with GSMA schema
		gsma_validated_body = gsma_schemas.Artefact.model_validate(request_body)
		body = gsma_validated_body.model_dump()
		except ValidationError as e:
		log.error(f"Invalid GSMA artefact request body: {e}")
		raise

		try:
		artefact_id = body["artefactId"]
		artefact_name = body["artefactName"]
		repo_data = body["artefactRepoLocation"]

		transformed = {
		"artefact_id": artefact_id,
		"artefact_name": artefact_name,
		"repo_name": repo_data.get("repoName", ""),
		"repo_type": request_body.get("repoType"),
		"repo_type": body.get("repoType"),
		"repo_url": repo_data["repoURL"],
		"user_name": repo_data.get("userName"),
		"password": repo_data.get("password"),
		@@ -855,6 +879,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		request=response.request,
		)
		return response

		except I2EdgeError as e:
		log.error(f"Failed to create artefact: {e}")
		raise
		@@ -870,7 +895,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		response = self.get_artefact(artefact_id)
		if response.status_code == 200:
		response_json = response.json()
		content = gsma_schemas.Artefact(
		content = gsma_schemas.ArtefactRetrieve(
		artefactId=response_json.get("artefact_id"),
		appProviderId=response_json.get("id"),
		artefactName=response_json.get("name"),
		@@ -889,7 +914,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		),
		)
		try:
		validated_data = gsma_schemas.Artefact.model_validate(content)
		validated_data = gsma_schemas.ArtefactRetrieve.model_validate(content)
		except ValidationError as e:
		raise ValueError(f"Invalid schema: {e}")
		return build_custom_http_response(
		@@ -940,13 +965,18 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param request_body: Payload with onboarding info.
		:return: Response with onboarding confirmation.
		"""
		body = deepcopy(request_body)
		try:
		body["app_id"] = body.pop("appId")
		body.pop("edgeAppFQDN", None)
		data = body
		# Validate input against GSMA schema
		gsma_validated_body = gsma_schemas.AppOnboardManifestGSMA.model_validate(request_body)
		data = gsma_validated_body.model_dump()
		except ValidationError as e:
		log.error(f"Invalid GSMA input schema: {e}")
		raise
		try:
		data["app_id"] = data.pop("appId")
		data.pop("edgeAppFQDN", None)
		payload = i2edge_schemas.ApplicationOnboardingRequest(profile_data=data)
		url = "{}/application/onboarding".format(self.base_url)
		url = f"{self.base_url}/application/onboarding"
		response = i2edge_post(url, payload, expected_status=201)
		return build_custom_http_response(
		status_code=200,
		@@ -1029,18 +1059,26 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param request_body: Payload with updated onboarding info.
		:return: Response with update confirmation.
		"""
		try:
		# Validate input body using GSMA schema
		gsma_validated_body = gsma_schemas.PatchOnboardedAppGSMA.model_validate(request_body)
		patch_payload = gsma_validated_body.model_dump()
		except ValidationError as e:
		log.error(f"Invalid GSMA input schema: {e}")
		raise
		try:
		url = f"{self.base_url}/application/onboarding/{app_id}"
		params = {}
		response = i2edge_get(url, params, expected_status=200)
		response_json = response.json()
		app_component_specs = request_body.get("appComponentSpecs")
		app_qos_profile = request_body.get("appUpdQoSProfile")
		# Update fields
		app_component_specs = patch_payload.get("appComponentSpecs")
		app_qos_profile = patch_payload.get("appUpdQoSProfile")
		response_json["profile_data"]["appQoSProfile"] = app_qos_profile
		response_json["profile_data"]["appComponentSpecs"] = app_component_specs
		data = response_json.get("profile_data")
		try:
		payload = i2edge_schemas.ApplicationOnboardingRequest(profile_data=data)
		url = "{}/application/onboarding/{}".format(self.base_url, app_id)
		url = f"{self.base_url}/application/onboarding/{app_id}"
		response = i2edge_patch(url, payload, expected_status=200)
		return build_custom_http_response(
		status_code=200,
		@@ -1088,10 +1126,16 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param request_body: Payload with deployment information.
		:return: Response with deployment details.
		"""
		body = deepcopy(request_body)
		try:
		zone_id = body.get("zoneInfo").get("zoneId")
		flavour_id = body.get("zoneInfo").get("flavourId")
		# Validate input against GSMA schema
		gsma_validated_body = gsma_schemas.AppDeployPayloadGSMA.model_validate(request_body)
		body = gsma_validated_body.model_dump()
		except ValidationError as e:
		log.error(f"Invalid GSMA input schema: {e}")
		raise
		try:
		zone_id = body.get("zoneInfo", {}).get("zoneId")
		flavour_id = body.get("zoneInfo", {}).get("flavourId")
		app_deploy_data = i2edge_schemas.AppDeployData(
		appId=body.get("appId"),
		appProviderId=body.get("appProviderId"),
		@@ -1099,18 +1143,15 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		zoneInfo=i2edge_schemas.ZoneInfoRef(flavourId=flavour_id, zoneId=zone_id),
		)
		payload = i2edge_schemas.AppDeploy(app_deploy_data=app_deploy_data)
		url = "{}/application_instance".format(self.base_url)
		url = f"{self.base_url}/application_instance"
		response = i2edge_post(url, payload, expected_status=202)

		response_json = response.json()
		content = gsma_schemas.AppInstance(
		# Validate response against GSMA schema
		app_instance = gsma_schemas.AppInstance(
		zoneId=response_json.get("zoneID"),
		appInstIdentifier=response_json.get("app_instance_id"),
		)
		try:
		validated_data = gsma_schemas.AppInstance.model_validate(content)
		except ValidationError as e:
		raise ValueError(f"Invalid schema: {e}")
		validated_data = gsma_schemas.AppInstance.model_validate(app_instance)
		return build_custom_http_response(
		status_code=202,
		content=validated_data.model_dump(),
		@@ -1123,6 +1164,10 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		log.error(f"Failed to deploy app: {e}")
		raise

		except ValidationError as e:
		log.error(f"Invalid GSMA response schema: {e}")
		raise

		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.

src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py

+30 −1

Original line number	Diff line number	Diff line
		@@ -128,7 +128,20 @@ class ArtefactRepoLocation(BaseModel):
		token: Optional[str] = None


		class Artefact(BaseModel):
		class ArtefactComponentSpec(BaseModel):
		componentName: str
		images: List[str]
		numOfInstances: int
		restartPolicy: str
		commandLineParams: Optional[dict] = None
		exposedInterfaces: Optional[List[dict]] = None
		computeResourceProfile: Optional[dict] = None
		compEnvParams: Optional[List[dict]] = None
		deploymentConfig: Optional[dict] = None
		persistentVolumes: Optional[List[dict]] = None


		class ArtefactRetrieve(BaseModel):
		artefactId: str
		appProviderId: Optional[str] = None
		artefactName: str
		@@ -142,6 +155,22 @@ class Artefact(BaseModel):
		artefactRepoLocation: Optional[ArtefactRepoLocation] = None


		class Artefact(BaseModel):
		artefactId: str
		appProviderId: str
		artefactName: str
		artefactVersionInfo: str
		artefactDescription: Optional[str] = None
		artefactVirtType: Literal["VM_TYPE", "CONTAINER_TYPE"]
		artefactFileName: Optional[str] = None
		artefactFileFormat: Optional[Literal["ZIP", "TAR", "TEXT", "TARGZ"]] = None
		artefactDescriptorType: Literal["HELM", "TERRAFORM", "ANSIBLE", "SHELL", "COMPONENTSPEC"]
		repoType: Optional[Literal["PRIVATEREPO", "PUBLICREPO", "UPLOAD"]] = None
		artefactRepoLocation: Optional[ArtefactRepoLocation] = None
		artefactFile: Optional[str] = None
		componentSpec: List[ArtefactComponentSpec]


		# ---------------------------
		# ApplicationOnboardingManagement
		# ---------------------------

Admin message