feature/add-gsma-schemas-to-edgecloud-adapters: add schemas for onboard, artefact and deploy (2ab09603) · Commits · OOP / code / transformation-function-sdk

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

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -549,7 +549,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	"""		"""
	pass		pass

	def get_all_deployed_apps_gsma(self, app_id: str, app_provider: str) -> List:		def get_all_deployed_apps_gsma(self) -> Response:
	"""		"""
	Retrieves all instances for a given application of partner OP		Retrieves all instances for a given application of partner OP

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

+235 −195

Original line number	Original line	Diff line number	Diff line
	@@ -735,25 +735,24 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url = f"{self.base_url}/zones/list"		url = f"{self.base_url}/zones/list"
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	try:		try:
	validated_data = gsma_schemas.ZonesList.model_validate(response_json)		validated_data = gsma_schemas.ZonesList.model_validate(response_json)
	except ValidationError as e:		except ValidationError as e:
	raise ValueError(f"Response from /zones/list is not a valid schema: {e}")		raise ValueError(f"Invalid schema: {e}")

	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=[zone.model_dump() for zone in validated_data.root],		content=[zone.model_dump_json() for zone in validated_data.root],
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		log.error(f"Failed to obtain Zones list from i2edge: {e}")
			raise

	def get_edge_cloud_zones_gsma(self) -> Response:		def get_edge_cloud_zones_gsma(self) -> Response:
	"""		"""
	@@ -764,25 +763,24 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url = f"{self.base_url}/zones"		url = f"{self.base_url}/zones"
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	mapped = [map_zone(zone) for zone in response_json]		mapped = [map_zone(zone) for zone in response_json]
	try:		try:
	validated_data = gsma_schemas.ZoneRegisteredDataList.model_validate(mapped)		validated_data = gsma_schemas.ZoneRegisteredDataList.model_validate(mapped)
	except ValidationError as e:		except ValidationError as e:
	raise ValueError(f"Invalid response schema from /zones: {e}")		raise ValueError(f"Invalid schema {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=validated_data.model_dump(),		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		log.error(f"Failed to obtain Zones details from i2edge: {e}")
			raise

	def get_edge_cloud_zone_details_gsma(self, zone_id: str) -> Response:		def get_edge_cloud_zone_details_gsma(self, zone_id: str) -> Response:
	"""		"""
	@@ -795,25 +793,24 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url = f"{self.base_url}/zone/{zone_id}"		url = f"{self.base_url}/zone/{zone_id}"
	params = {}		params = {}
	try:		try:
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	mapped = map_zone(response_json)		mapped = map_zone(response_json)
	try:		try:
	validated_data = gsma_schemas.ZoneRegisteredData.model_validate(mapped)		validated_data = gsma_schemas.ZoneRegisteredData.model_validate(mapped)
	except ValidationError as e:		except ValidationError as e:
	raise ValueError(f"Invalid response schema from /zones/{zone_id}: {e}")		raise ValueError(f"Invalid schema: {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=validated_data.model_dump(),		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response
	except I2EdgeError as e:		except I2EdgeError as e:
	raise e		log.error(f"Failed to obtain Zones details from i2edge: {e}")
			raise

	# ------------------------------------------------------------------------		# ------------------------------------------------------------------------
	# Artefact Management (GSMA)		# Artefact Management (GSMA)
	@@ -836,8 +833,8 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	transformed = {		transformed = {
	"artefact_id": artefact_id,		"artefact_id": artefact_id,
	"artefact_name": artefact_name,		"artefact_name": artefact_name,
	"repo_name": repo_data.get("repoName", "unknown-repo"),		"repo_name": repo_data.get("repoName"),
	"repo_type": request_body.get("repoType", "PUBLICREPO"),		"repo_type": request_body.get("repoType"),
	"repo_url": repo_data["repoURL"],		"repo_url": repo_data["repoURL"],
	"user_name": repo_data.get("userName"),		"user_name": repo_data.get("userName"),
	"password": repo_data.get("password"),		"password": repo_data.get("password"),
	@@ -855,8 +852,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	request=response.request,		request=response.request,
	)		)
	return response		return response
	except KeyError as e:		except I2EdgeError as e:
	raise I2EdgeError(f"Missing required field in GSMA artefact payload: {e}")		log.error(f"Failed to create artefact: {e}")
			raise

	def get_artefact_gsma(self, artefact_id: str) -> Response:		def get_artefact_gsma(self, artefact_id: str) -> Response:
	"""		"""
	@@ -869,36 +867,40 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	response = self.get_artefact(artefact_id)		response = self.get_artefact(artefact_id)
	if response.status_code == 200:		if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	print(response_json)		content = gsma_schemas.Artefact(
	content = {		artefactId=response_json.get("artefact_id"),
	"artefactId": response_json.get("artefact_id"),		appProviderId=response_json.get("id"),
	"appProviderId": "Ihs0gCqO65SHTz",		artefactName=response_json.get("name"),
	"artefactName": response_json.get("name"),		artefactDescription="Description",
	"artefactDescription": "string",		artefactVersionInfo=response_json.get("version"),
	"artefactVersionInfo": response_json.get("version"),		artefactVirtType="VM_TYPE",
	"artefactVirtType": "VM_TYPE",		artefactFileName="FileName",
	"artefactFileName": "stringst",		artefactFileFormat="TAR",
	"artefactFileFormat": "ZIP",		artefactDescriptorType="HELM",
	"artefactDescriptorType": "HELM",		repoType=response_json.get("repo_type"),
	"repoType": response_json.get("repo_type"),		artefactRepoLocation=gsma_schemas.ArtefactRepoLocation(
	"artefactRepoLocation": {		repoURL=response_json.get("repo_url"),
	"repoURL": response_json.get("repo_url"),		userName=response_json.get("repo_user_name"),
	"userName": response_json.get("repo_user_name"),		password=response_json.get("repo_password"),
	"password": response_json.get("repo_password"),		token=response_json.get("repo_token"),
	"token": response_json.get("repo_token"),		),
	},		)
	}		try:
			validated_data = gsma_schemas.Artefact.model_validate(content)
			except ValidationError as e:
			raise ValueError(f"Invalid schema: {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=content,		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		return response
	except KeyError as e:		except I2EdgeError as e:
	raise I2EdgeError(f"Missing artefactId in GSMA payload: {e}")		log.error(f"Failed to retrieve artefact: {e}")
			raise

	def delete_artefact_gsma(self, artefact_id: str) -> Response:		def delete_artefact_gsma(self, artefact_id: str) -> Response:
	"""		"""
	@@ -942,8 +944,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	data = body		data = body
	payload = i2edge_schemas.ApplicationOnboardingRequest(profile_data=data)		payload = i2edge_schemas.ApplicationOnboardingRequest(profile_data=data)
	url = "{}/application/onboarding".format(self.base_url)		url = "{}/application/onboarding".format(self.base_url)
	response = i2edge_post(url, payload)		response = i2edge_post(url, payload, expected_status=201)
	if response.status_code == 201:
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content={"response": "Application onboarded successfully"},		content={"response": "Application onboarded successfully"},
	@@ -952,9 +953,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		except I2EdgeError as e:
	except KeyError as e:		log.error(f"Failed to onboard app: {e}")
	raise I2EdgeError(f"Missing required field in GSMA onboarding payload: {e}")		raise

	def get_onboarded_app_gsma(self, app_id: str) -> Response:		def get_onboarded_app_gsma(self, app_id: str) -> Response:
	"""		"""
	@@ -963,30 +964,58 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	:param app_id: Identifier of the application onboarded.		:param app_id: Identifier of the application onboarded.
	:return: Response with application details.		:return: Response with application details.
	"""		"""
			url = f"{self.base_url}/application/onboarding/{app_id}"
			params = {}
	try:		try:
	response = self.get_onboarded_app(app_id)		response = i2edge_get(url, params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	profile_data = response_json.get("profile_data")		profile_data = response_json.get("profile_data")
	content = {		app_deployment_zones = profile_data.get("appDeploymentZones")
	"appId": profile_data.get("app_id"),		app_metadata = profile_data.get("appMetaData")
	"appProviderId": "string",		app_qos_profile = profile_data.get("appQoSProfile")
	"appDeploymentZones": profile_data.get("appDeploymentZones"),		app_component_specs = profile_data.get("appComponentSpecs")
	"appMetaData": profile_data.get("appMetadata"),		content = gsma_schemas.ApplicationModel(
	"appQoSProfile": profile_data.get("appQoSProfile"),		appId=profile_data.get("app_id"),
	"appComponentSpecs": profile_data.get("appComponentSpecs"),		appProviderId="from_FM",
	}		appDeploymentZones=[
			gsma_schemas.AppDeploymentZone(countryCode="ES", zoneInfo=zone_id)
			for zone_id in app_deployment_zones
			],
			appMetaData=gsma_schemas.AppMetaData(
			appName=app_metadata.get("appName"),
			version=app_metadata.get("version"),
			appDescription=app_metadata.get("appDescription"),
			mobilitySupport=app_metadata.get("mobilitySupport"),
			accessToken=app_metadata.get("accessToken"),
			category=app_metadata.get("category"),
			),
			appQoSProfile=gsma_schemas.AppQoSProfile(
			latencyConstraints=app_qos_profile.get("latencyConstraints"),
			bandwidthRequired=app_qos_profile.get("bandwidthRequired"),
			multiUserClients=app_qos_profile.get("multiUserClients"),
			noOfUsersPerAppInst=app_qos_profile.get("noOfUsersPerAppInst"),
			appProvisioning=app_qos_profile.get("appProvisioning"),
			),
			appComponentSpecs=[
			gsma_schemas.AppComponentSpec(**component) for component in app_component_specs
			],
			onboardStatusInfo="ONBOARDED",
			)
			try:
			validated_data = gsma_schemas.ApplicationModel.model_validate(content)
			except ValidationError as e:
			raise ValueError(f"Invalid schema: {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=content,		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		except I2EdgeError as e:
	except KeyError as e:		log.error(f"Failed to get onboarded app: {e}")
	raise I2EdgeError(f"Missing appId in GSMA payload: {e}")		raise

	def patch_onboarded_app_gsma(self, app_id: str, request_body: dict) -> Response:		def patch_onboarded_app_gsma(self, app_id: str, request_body: dict) -> Response:
	"""		"""
	@@ -1018,8 +1047,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	request=response.request,		request=response.request,
	)		)
	return response		return response
	except KeyError as e:		except I2EdgeError as e:
	raise I2EdgeError(f"Missing appId in GSMA payload: {e}")		log.error(f"Failed to delete onboarded app: {e}")
			raise

	# ------------------------------------------------------------------------		# ------------------------------------------------------------------------
	# Application Deployment Management (GSMA)		# Application Deployment Management (GSMA)
	@@ -1044,24 +1074,28 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	)		)
	payload = i2edge_schemas.AppDeploy(app_deploy_data=app_deploy_data)		payload = i2edge_schemas.AppDeploy(app_deploy_data=app_deploy_data)
	url = "{}/application_instance".format(self.base_url)		url = "{}/application_instance".format(self.base_url)
	response = i2edge_post(url, payload)		response = i2edge_post(url, payload, expected_status=202)
	if response.status_code == 202:
	response_json = response.json()		response_json = response.json()
	content = {		content = gsma_schemas.AppInstance(
	"zoneId": response_json.get("zoneID"),		zoneId=response_json.get("zoneID"),
	"appInstIdentifier": response_json.get("app_instance_id"),		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}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=202,		status_code=202,
	content=content,		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		except I2EdgeError as e:
	except KeyError as e:		log.error(f"Failed to deploy app: {e}")
	raise I2EdgeError(f"Missing required field in GSMA deployment payload: {e}")		raise

	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:
	"""		"""
	@@ -1075,26 +1109,31 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	try:		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 = {}		params = {}
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	content = {		content = gsma_schemas.AppInstanceStatus(
	"appInstanceState": response_json.get("appInstanceState"),		appInstanceState=response_json.get("appInstanceState"),
	"accesspointInfo": response_json.get("accesspointInfo"),		accesspointInfo=response_json.get("accesspointInfo"),
	}		)
			try:
			validated_data = gsma_schemas.AppInstanceStatus.model_validate(content)
			except ValidationError as e:
			raise ValueError(f"Invalid schema: {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=content,		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response
	except KeyError as e:
	raise I2EdgeError(f"Missing appId or zoneId in GSMA payload: {e}")

	def get_all_deployed_apps_gsma(self, app_id: str, app_provider: str) -> Response:		except I2EdgeError as e:
			log.error(f"Failed to retrieve deployed app: {e}")
			raise

			def get_all_deployed_apps_gsma(self) -> Response:
	"""		"""
	Retrieves all instances for a given application of partner OP using GSMA federation.		Retrieves all instances for a given application of partner OP using GSMA federation.

	@@ -1105,13 +1144,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	try:		try:
	url = "{}/application_instances".format(self.base_url)		url = "{}/application_instances".format(self.base_url)
	params = {}		params = {}
	response = i2edge_get(url, params=params)		response = i2edge_get(url, params=params, expected_status=200)
	if response.status_code == 200:
	response_json = response.json()		response_json = response.json()
	response_list = []		response_list = []
	for item in response_json:		for item in response_json:
	content = [		content = {
	{
	"zoneId": item.get("app_spec")		"zoneId": item.get("app_spec")
	.get("nodeSelector")		.get("nodeSelector")
	.get("feature.node.kubernetes.io/zoneID"),		.get("feature.node.kubernetes.io/zoneID"),
	@@ -1122,19 +1159,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	}		}
	],		],
	}		}
	]
	response_list.append(content)		response_list.append(content)
			try:
			validated_data = gsma_schemas.ZoneIdentifierList.model_validate(response_list)
			except ValidationError as e:
			raise ValueError(f"Invalid schema: {e}")
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content=response_list,		content=validated_data.model_dump_json(),
	headers={"Content-Type": self.content_type_gsma},		headers={"Content-Type": self.content_type_gsma},
	encoding=self.encoding_gsma,		encoding=self.encoding_gsma,
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		except I2EdgeError as e:
	except KeyError as e:		log.error(f"Failed to retrieve apps: {e}")
	raise I2EdgeError(f"Error retrieving apps: {e}")		raise

	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:
	"""		"""
	@@ -1147,8 +1188,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	"""		"""
	try:		try:
	url = "{}/application_instance".format(self.base_url)		url = "{}/application_instance".format(self.base_url)
	response = i2edge_delete(url, app_instance_id)		response = i2edge_delete(url, app_instance_id, expected_status=200)
	if response.status_code == 200:
	return build_custom_http_response(		return build_custom_http_response(
	status_code=200,		status_code=200,
	content={"response": "Application instance termination request accepted"},		content={"response": "Application instance termination request accepted"},
	@@ -1157,6 +1197,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	url=response.url,		url=response.url,
	request=response.request,		request=response.request,
	)		)
	return response		except I2EdgeError as e:
	except KeyError as e:		log.error(f"Failed to delete app: {e}")
	raise I2EdgeError(f"Missing appInstanceId in GSMA payload: {e}")		raise

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

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -389,7 +389,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	"""		"""
	pass		pass

	def get_all_deployed_apps_gsma(self, app_id: str, app_provider: str) -> List:		def get_all_deployed_apps_gsma(self) -> Response:
	"""		"""
	Retrieves all instances for a given application of partner OP		Retrieves all instances for a given application of partner OP

src/sunrise6g_opensdk/edgecloud/core/edgecloud_interface.py

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -278,7 +278,7 @@ class EdgeCloudManagementInterface(ABC):
	pass		pass

	@abstractmethod		@abstractmethod
	def get_all_deployed_apps_gsma(self, app_id: str, app_provider: str) -> Response:		def get_all_deployed_apps_gsma(self) -> Response:
	"""		"""
	Retrieves all instances for a given application of partner OP		Retrieves all instances for a given application of partner OP

src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py

+96 −1

Original line number	Original line	Diff line number	Diff line
	from typing import List, Literal, Optional		from typing import List, Literal, Optional

	from pydantic import BaseModel, Field, RootModel		from pydantic import BaseModel, Field, HttpUrl, RootModel

	# ---------------------------		# ---------------------------
	# FederationManagement		# FederationManagement
	@@ -121,11 +121,106 @@ class ZoneRegisteredDataList(RootModel[List[ZoneRegisteredData]]):
	# ---------------------------		# ---------------------------


			class ArtefactRepoLocation(BaseModel):
			repoURL: HttpUrl
			userName: Optional[str] = None
			password: Optional[str] = None
			token: Optional[str] = None


			class Artefact(BaseModel):
			artefactId: str
			appProviderId: str = None
			artefactName: str
			artefactDescription: Optional[str] = None
			artefactVersionInfo: str
			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


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


			class AppDeploymentZone(BaseModel):
			countryCode: str
			zoneInfo: str


			class AppMetaData(BaseModel):
			appName: str
			version: str
			appDescription: Optional[str] = None
			mobilitySupport: bool = False
			accessToken: str
			category: Optional[
			Literal[
			"IOT",
			"HEALTH_CARE",
			"GAMING",
			"VIRTUAL_REALITY",
			"SOCIALIZING",
			"SURVEILLANCE",
			"ENTERTAINMENT",
			"CONNECTIVITY",
			"PRODUCTIVITY",
			"SECURITY",
			"INDUSTRIAL",
			"EDUCATION",
			"OTHERS",
			]
			] = None


			class AppQoSProfile(BaseModel):
			latencyConstraints: Literal["NONE", "LOW", "ULTRALOW"]
			bandwidthRequired: int = Field(..., ge=1)
			multiUserClients: Literal["APP_TYPE_SINGLE_USER", "APP_TYPE_MULTI_USER"]
			noOfUsersPerAppInst: int = 1
			appProvisioning: bool = True


			class AppComponentSpec(BaseModel):
			serviceNameNB: str
			serviceNameEW: str
			componentName: str
			artefactId: str


			class ApplicationModel(BaseModel):
			appId: str
			appProviderId: str
			appDeploymentZones: List[AppDeploymentZone] = Field(..., min_length=1)
			appMetaData: AppMetaData
			appQoSProfile: AppQoSProfile
			appComponentSpecs: List[AppComponentSpec] = Field(..., min_length=1)
			onboardStatusInfo: Literal["PENDING", "ONBOARDED", "DEBOARDING", "REMOVED", "FAILED"]


	# ---------------------------		# ---------------------------
	# ApplicationDeploymentManagement		# ApplicationDeploymentManagement
	# ---------------------------		# ---------------------------


			class AppInstance(BaseModel):
			zoneId: str
			appInstIdentifier: str


			class AppInstanceStatus(BaseModel):
			appInstanceState: Literal["PENDING", "READY", "FAILED", "TERMINATING", "DEPLOYED"]
			accesspointInfo: List[dict]


			class ZoneIdentifier(BaseModel):
			zoneId: str
			appInstanceInfo: List[dict]


			class ZoneIdentifierList(RootModel[List[ZoneIdentifier]]):
			pass