Merge pull request #143 from SunriseOpenOperatorPlatform/feature/add-edgecloud-gsma-i2edge (457d1d01) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/common/adapters_schemas.py

0 → 100644

+28 −0

Original line number	Diff line number	Diff line
		# EXAMPLE OF PYDANTIC SCHEMAS
		from typing import Optional

		from pydantic import AnyHttpUrl, BaseModel, Field


		class _AdapterBase(BaseModel):
		client_name: str = Field(..., min_length=1)
		base_url: AnyHttpUrl

		class Config:
		extra = "allow"


		class EdgeCloudConfig(_AdapterBase):
		pass


		class NetworkConfig(_AdapterBase):
		pass


		class AdapterSpecs(BaseModel):
		edgecloud: Optional[EdgeCloudConfig] = None
		network: Optional[NetworkConfig] = None

		class Config:
		extra = "forbid"

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

+201 −0

Original line number	Diff line number	Diff line
		@@ -431,3 +431,204 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"flavoursSupported": flavours_supported,
		}
		return result

		# --- GSMA-specific methods ---

		# FederationManagement

		def get_edge_cloud_zones_list_gsma(self) -> List:
		"""
		Retrieves details of all Zones

		:return: List.
		"""
		pass

		def get_edge_cloud_zones_gsma(self, federation_context_id: str) -> List:
		"""
		Retrieves details of Zones

		:param federation_context_id: Identifier of the federation context.
		:return: List.
		"""
		pass

		# AvailabilityZoneInfoSynchronization

		def availability_zone_info_gsma(
		self, federation_context_id: str, request_body: dict
		) -> Dict:
		"""
		Originating OP informs partner OP that it is willing to access
		the specified zones and partner OP shall reserve compute and
		network resources for these zones.

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload.
		:return:
		"""
		pass

		def get_edge_cloud_zone_details_gsma(
		self, federation_context_id: str, zone_id: str
		) -> Dict:
		"""
		Retrieves details of a specific Edge Cloud Zone reserved
		for the specified zone by the partner OP.

		:param federation_context_id: Identifier of the federation context.
		:param zone_id: Unique identifier of the Edge Cloud Zone.
		:return: Dictionary with Edge Cloud Zone details.
		"""
		pass

		# ArtefactManagement

		def create_artefact_gsma(
		self, federation_context_id: str, request_body: dict
		) -> Dict:
		"""
		Uploads application artefact on partner OP. Artefact is a zip file
		containing scripts and/or packaging files like Terraform or Helm
		which are required to create an instance of an application

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload with artefact information.
		:return: Details with artefact deployment info.
		"""
		pass

		def get_artefact_gsma(self, federation_context_id: str, artefact_id: str) -> Dict:
		"""
		Retrieves details about an artefact

		:param federation_context_id: Identifier of the federation context.
		:param artefact_id: Unique identifier of the artefact.
		:return: Dictionary with artefact details.
		"""
		pass

		def delete_artefact_gsma(
		self, federation_context_id: str, artefact_id: str
		) -> Dict:
		"""
		Removes an artefact from partners OP.

		:param federation_context_id: Identifier of the federation context.
		:param artefact_id: Unique identifier of the artefact.
		:return: Dictionary with artefact deletion details.
		"""
		pass

		# ApplicationOnboardingManagement

		def onboard_app_gsma(self, federation_context_id: str, request_body: dict) -> Dict:
		"""
		Submits an application details to a partner OP.
		Based on the details provided, partner OP shall do bookkeeping,
		resource validation and other pre-deployment operations.

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload with onboarding info.
		:return: Dictionary with onboarding details.
		"""
		pass

		def get_onboarded_app_gsma(self, federation_context_id: str, app_id: str) -> Dict:
		"""
		Retrieves application details from partner OP

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		def patch_onboarded_app_gsma(
		self, federation_context_id: str, app_id: str, request_body: dict
		) -> Dict:
		"""
		Updates partner OP about changes in application compute resource requirements,
		QOS Profile, associated descriptor or change in associated components

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		def delete_onboarded_app_gsma(self, federation_context_id: str, app_id: str):
		"""
		Deboards an application from specific partner OP zones

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		# ApplicationDeploymentManagement

		def deploy_app_gsma(
		self, federation_context_id: str, idempotency_key: str, request_body: dict
		):
		"""
		Instantiates an application on a partner OP zone.

		:param federation_context_id: Identifier of the federation context.
		:param idempotency_key: Idempotency key.
		:return:
		"""
		pass

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

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_instance_id: Identifier of the deployed app.
		:param zone_id: Identifier of the zone
		:return:
		"""
		pass

		def get_all_deployed_apps_gsma(
		self,
		federation_context_id: str,
		app_id: str,
		app_provider: str,
		):
		"""
		Retrieves all application instance of partner OP

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_provider: App provider
		:return:
		"""
		pass

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

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_instance_id: Identifier of the deployed app.
		:param zone_id: Identifier of the zone
		:return:
		"""
		pass

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

+460 −4

File changed.

Preview size limit exceeded, changes collapsed.

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

+22 −4

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@
		#
		# Contributors:
		# - Sergio Giménez (sergio.gimenez@i2cat.net)
		# - César Cajas (cesar.cajas@i2cat.net)
		##
		import json
		from typing import Optional
		@@ -45,7 +46,7 @@ def i2edge_post(url: str, model_payload: BaseModel) -> dict:
		try:
		response = requests.post(url, data=json_payload, headers=headers)
		response.raise_for_status()
		return response.json()
		return response
		except requests.exceptions.HTTPError as e:
		i2edge_err_msg = get_error_message_from(response)
		err_msg = "Failed to deploy app: {}. Detail: {}".format(i2edge_err_msg, e)
		@@ -53,6 +54,23 @@ def i2edge_post(url: str, model_payload: BaseModel) -> dict:
		raise I2EdgeError(err_msg)


		def i2edge_put(url: str, model_payload: BaseModel) -> dict:
		headers = {
		"Content-Type": "application/json",
		"accept": "application/json",
		}
		json_payload = json.dumps(model_payload.model_dump(mode="json"))
		try:
		response = requests.put(url, data=json_payload, headers=headers)
		response.raise_for_status()
		return response
		except requests.exceptions.HTTPError as e:
		i2edge_err_msg = get_error_message_from(response)
		err_msg = "Failed to patch: {}. Detail: {}".format(i2edge_err_msg, e)
		log.error(err_msg)
		raise I2EdgeError(err_msg)


		def i2edge_post_multiform_data(url: str, model_payload: BaseModel) -> dict:
		headers = {
		"accept": "application/json",
		@@ -62,7 +80,7 @@ def i2edge_post_multiform_data(url: str, model_payload: BaseModel) -> dict:
		try:
		response = requests.post(url, data=payload_in_str, headers=headers)
		response.raise_for_status()
		return response.json()
		return response
		except requests.exceptions.HTTPError as e:
		i2edge_err_msg = get_error_message_from(response)
		err_msg = "Failed to deploy app: {}. Detail: {}".format(i2edge_err_msg, e)
		@@ -76,7 +94,7 @@ def i2edge_delete(url: str, id: str) -> dict:
		query = "{}/{}".format(url, id)
		response = requests.delete(query, headers=headers)
		response.raise_for_status()
		return response.json()
		return response
		except requests.exceptions.HTTPError as e:
		i2edge_err_msg = get_error_message_from(response)
		err_msg = "Failed to undeploy app: {}. Detail: {}".format(i2edge_err_msg, e)
		@@ -89,7 +107,7 @@ def i2edge_get(url: str, params: Optional[dict]):
		try:
		response = requests.get(url, params=params, headers=headers)
		response.raise_for_status()
		return response.json()
		return response
		except requests.exceptions.HTTPError as e:
		i2edge_err_msg = get_error_message_from(response)
		err_msg = "Failed to get apps: {}. Detail: {}".format(i2edge_err_msg, e)

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

+202 −0

Original line number	Diff line number	Diff line
		@@ -263,3 +263,205 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		}
		]
		return app

		# --- GSMA-specific methods ---

		# FederationManagement

		def get_edge_cloud_zones_list_gsma(self) -> List:
		"""
		Retrieves details of all Zones

		:return: List.
		"""

		pass

		def get_edge_cloud_zones_gsma(self, federation_context_id: str) -> List:
		"""
		Retrieves details of Zones

		:param federation_context_id: Identifier of the federation context.
		:return: List.
		"""
		pass

		# AvailabilityZoneInfoSynchronization

		def availability_zone_info_gsma(
		self, federation_context_id: str, request_body: dict
		) -> Dict:
		"""
		Originating OP informs partner OP that it is willing to access
		the specified zones and partner OP shall reserve compute and
		network resources for these zones.

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload.
		:return:
		"""
		pass

		def get_edge_cloud_zone_details_gsma(
		self, federation_context_id: str, zone_id: str
		) -> Dict:
		"""
		Retrieves details of a specific Edge Cloud Zone reserved
		for the specified zone by the partner OP.

		:param federation_context_id: Identifier of the federation context.
		:param zone_id: Unique identifier of the Edge Cloud Zone.
		:return: Dictionary with Edge Cloud Zone details.
		"""
		pass

		# ArtefactManagement

		def create_artefact_gsma(
		self, federation_context_id: str, request_body: dict
		) -> Dict:
		"""
		Uploads application artefact on partner OP. Artefact is a zip file
		containing scripts and/or packaging files like Terraform or Helm
		which are required to create an instance of an application

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload with artefact information.
		:return: Details with artefact deployment info.
		"""
		pass

		def get_artefact_gsma(self, federation_context_id: str, artefact_id: str) -> Dict:
		"""
		Retrieves details about an artefact

		:param federation_context_id: Identifier of the federation context.
		:param artefact_id: Unique identifier of the artefact.
		:return: Dictionary with artefact details.
		"""
		pass

		def delete_artefact_gsma(
		self, federation_context_id: str, artefact_id: str
		) -> Dict:
		"""
		Removes an artefact from partners OP.

		:param federation_context_id: Identifier of the federation context.
		:param artefact_id: Unique identifier of the artefact.
		:return: Dictionary with artefact deletion details.
		"""
		pass

		# ApplicationOnboardingManagement

		def onboard_app_gsma(self, federation_context_id: str, request_body: dict) -> Dict:
		"""
		Submits an application details to a partner OP.
		Based on the details provided, partner OP shall do bookkeeping,
		resource validation and other pre-deployment operations.

		:param federation_context_id: Identifier of the federation context.
		:param request_body: Payload with onboarding info.
		:return: Dictionary with onboarding details.
		"""
		pass

		def get_onboarded_app_gsma(self, federation_context_id: str, app_id: str) -> Dict:
		"""
		Retrieves application details from partner OP

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		def patch_onboarded_app_gsma(
		self, federation_context_id: str, app_id: str, request_body: dict
		) -> Dict:
		"""
		Updates partner OP about changes in application compute resource requirements,
		QOS Profile, associated descriptor or change in associated components

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		def delete_onboarded_app_gsma(self, federation_context_id: str, app_id: str):
		"""
		Deboards an application from specific partner OP zones

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass

		# ApplicationDeploymentManagement

		def deploy_app_gsma(
		self, federation_context_id: str, idempotency_key: str, request_body: dict
		):
		"""
		Instantiates an application on a partner OP zone.

		:param federation_context_id: Identifier of the federation context.
		:param idempotency_key: Idempotency key.
		:return:
		"""
		pass

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

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_instance_id: Identifier of the deployed app.
		:param zone_id: Identifier of the zone
		:return:
		"""
		pass

		def get_all_deployed_apps_gsma(
		self,
		federation_context_id: str,
		app_id: str,
		app_provider: str,
		):
		"""
		Retrieves all application instance of partner OP

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_provider: App provider
		:return:
		"""
		pass

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

		:param federation_context_id: Identifier of the federation context.
		:param app_id: Identifier of the app.
		:param app_instance_id: Identifier of the deployed app.
		:param zone_id: Identifier of the zone
		:return:
		"""
		pass