feat: transformation/translation function for TMF921 for submitting application metadata (ebd25f26) · Commits · OOP / code / transformation-function-sdk

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

+128 −1

Original line number	Diff line number	Diff line
		# Mocked API for testing purposes
		import logging
		from decimal import Decimal
		from typing import Dict, List, Optional

		from kubernetes.client import V1Deployment
		import requests
		from bson.decimal128 import Decimal128
		from kubernetes.client import V1Deployment, V1Service
		from rdflib import RDF, Graph, Namespace, term
		from requests import Response

		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.core.piedge_encoder import (
		@@ -65,6 +69,129 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		request=None
		)

		def _intent_to_dict(self, intent):
		ICM = Namespace("http://tio.models.tmforum.org/tio/v3.6.0/IntentCommonModel/")
		LOG = Namespace("http://tio.models.tmforum.org/tio/v3.6.0/LogicalOperators/")
		g = Graph()
		g.parse(data=intent, format="turtle")
		decoded_intent = {}
		for expectation in g.subjects(RDF.type, ICM.PropertyExpectation):
		for blank_node in g.objects(expectation, LOG.allOf):
		prop = ""
		for inner_blank_node in g.objects(blank_node, LOG.equals):
		if type(inner_blank_node) is term.BNode:
		for _prop in g.objects(
		inner_blank_node, ICM.valuesOfTargetProperties
		):
		prop = _prop.fragment
		else:
		if isinstance(inner_blank_node.value, Decimal):
		decoded_intent[prop] = Decimal128(inner_blank_node.value)
		else:
		decoded_intent[prop] = inner_blank_node.value
		return decoded_intent


		def onboard_intent(self, intent):
		ttl_string = intent["expression"]["expressionValue"]
		decoded_intent = self._intent_to_dict(intent=ttl_string)

		body = {
		"appId": decoded_intent["appId"],
		"name": decoded_intent["app"],
		"appRepo": {"imagePath": decoded_intent["containerImage"]},
		"packageType": "UNDEFINED",
		"componentSpec": [
		{"networkInterfaces": [{"port": decoded_intent["containerPort"]}]}
		],
		"appProvider": "6g-intense-dmo",
		"requiredResources": {
		"applicationResources": {
		"cpuPool": {
		"numCPU": decoded_intent["cpuRequest"][:-1],
		"memory": decoded_intent["memoryRequest"][:-2],
		}
		}
		},
		"version": ""
		}

		onboarding = self.onboard_app(body)
		if onboarding.status_code != 201:
		return onboarding

		intent["_id"] = decoded_intent["appId"]
		try:
		saving = self.connector_db.insert_document_intent(intent)
		except Exception as exc:
		self.delete_onboarded_app(decoded_intent["appId"])
		return {
		"status": 500,
		"code": "INTERNAL",
		"message": "Internal server error: " + exc.args,
		}
		if saving == 200:
		return build_custom_http_response(
		status_code=200,
		content={"message": "Intent OK!",
		"id": decoded_intent["appId"]
		},
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)
		else:
		self.delete_onboarded_app(decoded_intent["appId"])
		return build_custom_http_response(
		status_code=saving,
		content={"message": "Intent Saving failed."},
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		def get_intents(self, fields, offset, limit):
		try:
		return self.connector_db.get_intents(fields, offset, limit)
		except Exception as exc:
		return {
		"status": 500,
		"code": "INTERNAL",
		"message": "Internal server error: " + exc.args,
		}

		def delete_intent(self, id):
		app_manifest = self.get_onboarded_app(id)
		res = self.delete_onboarded_app(id)
		if res.status_code != 204:
		return res

		try:
		intent_res = self.connector_db.delete_document_intent(id)
		except Exception as e:
		self.onboard_app(app_manifest)
		return {
		"status": 500,
		"code": "INTERNAL",
		"message": "Internal server error: " + e.args,
		}

		content = {}
		if intent_res == 200:
		content = {"message": "Intent successfully delete."}
		else:
		content = {"message": "Intent deletion failed"}

		return build_custom_http_response(
		status_code=intent_res,
		content=content,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		def onboard_app(self, app_manifest: AppManifest) -> Response:
		print(f"Submitting application: {app_manifest}")

src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/lib/utils/connector_db.py

+52 −0

Original line number	Diff line number	Diff line
		@@ -270,6 +270,58 @@ class ConnectorDB:
		except Exception as ce_:
		raise Exception("An exception occurred :", ce_)

		def insert_document_intent(self, intent=None):
		collection = "intents"
		myclient = pymongo.MongoClient(self._storage_url)
		mydbmongo = myclient[self.mydb_mongo]
		mycol = mydbmongo[collection]

		myquery = {"_id": intent["_id"]}
		mydoc = mycol.find_one(myquery)
		# keeps the last record (contains registrationStatus)
		if mydoc is not None:
		return 409
		try:
		mycol.insert_one(intent)
		return 200
		except Exception as ce_:
		raise Exception("An exception occurred :", ce_)


		def get_intents(self, fields, offset, limit):
		myclient = pymongo.MongoClient(self._storage_url)
		mydbmongo = myclient[self.mydb_mongo]
		mycol = mydbmongo["intents"]

		if type(fields) is str:
		fields = fields.split(",")

		try:
		cursor = mycol.find(projection=fields, skip=offset, limit=limit)
		return list(cursor)
		except Exception as e:
		raise Exception("An exception occured: ", e)


		def delete_document_intent(self, id):
		myclient = pymongo.MongoClient(self._storage_url)
		mydbmongo = myclient[self.mydb_mongo]
		mycol = mydbmongo["intents"]

		myquery = {"_id": id}
		mydoc = mycol.find_one(myquery)

		# keeps the last record (contains registrationStatus)
		if mydoc is None:
		return 404
		# raise Exception("Not found: PaaS name", document["paas_name"])
		try:
		mycol.delete_one(myquery)
		return 200
		except Exception as ce_:
		raise Exception("An exception occurred :", ce_)


		def get_documents_from_collection(
		self, collection_input, input_type=None, input_value=None
		) -> List[dict]:

Admin message