Adapts application onboarding management API (9f11e8e9) · Commits · OOP / code / federation-manager

src/adapters/fm_adapter/application_onboarding_management.py

0 → 100644

+395 −0

Original line number	Diff line number	Diff line
		# -------------------------------------------------------------------------- #
		# Copyright 2025-present, Federation Manager, by Software Networks, i2CAT #
		# #
		# Licensed under the Apache License, Version 2.0 (the "License"); you may #
		# not use this file except in compliance with the License. You may obtain #
		# a copy of the License at #
		# #
		# http://www.apache.org/licenses/LICENSE-2.0 #
		# #
		# Unless required by applicable law or agreed to in writing, software #
		# distributed under the License is distributed on an "AS IS" BASIS, #
		# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
		# See the License for the specific language governing permissions and #
		# limitations under the License. #
		# -------------------------------------------------------------------------- #
		import json
		from mongoengine.errors import ValidationError

		from models.mongo_document import OriginatingOperatorPlatformOriginatingOP
		from models.mongo_document import OriginatingApplicationOnboardingManagementOriginatingOP
		from models.mongo_document import OriginatingApplicationOnboardingManagementUpdateOriginatingOP
		from adapters.error import APIError
		from clients import fed_manager as fm_client


		def verify_required_header(partner_api_root):
		if not partner_api_root:
		raise APIError(400, "X-Partner-API-Root header is missing")


		def delete_app(federation_context_id, app_id, bearer_token, partner_api_root): # noqa: E501
		"""Deboards the application from any zones, if any, and deletes the App.

		# noqa: E501

		:param federation_context_id:
		:type federation_context_id: dict \| bytes
		:param app_id:
		:type app_id: dict \| bytes

		:rtype: None
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		try:
		# Find federation at originating OP
		originating_op_objects = OriginatingOperatorPlatformOriginatingOP.objects(id=federation_context_id)
		if not originating_op_objects:
		raise APIError(404, "Federation not found")
		originating_op_instance = originating_op_objects[0]

		# Find application onboarding at partner
		response_get = fm_client.get_profile(originating_op_instance.partner_federation_id, app_id, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_get and "status_code" in response_get:
		status_code = response_get["status_code"]
		error_message = response_get["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "appId" in response_get:
		originating_ao_objects = find_application_onboarding_at_originating_op(federation_context_id, app_id)
		if originating_ao_objects:
		# Delete application onboarding from partner
		response_data = fm_client.delete_profile(originating_op_instance.partner_federation_id, app_id,
		bearer_token, partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_data and "status_code" in response_data:
		status_code = response_data["status_code"]
		error_message = response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "deletion" in response_data:
		# If application onboarding has been removed well, remove application onboarding from originating
		delete_application_onboarding_originating_op(originating_ao_objects)
		return response_data, 200
		else:
		raise APIError(422, f"Unexpected response from partner API: {response_data}")
		else:
		raise APIError(409, "Application Onboarding exist at partner operator but not in originating operator")
		else:
		if find_application_onboarding_at_originating_op(federation_context_id, app_id):
		raise APIError(409, "Application Onboarding exist in originating operator but not at partner operator")
		else:
		raise APIError(404, "Application Onboarding not found")
		except ValidationError:
		raise APIError(400, f"Invalid federation context ID or app ID format: {federation_context_id}, {app_id}")
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while deleting application onboarding. Reason: {error}")


		def onboard_application(body, federation_context_id, bearer_token, partner_api_root): # noqa: E501
		"""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.

		# noqa: E501

		:param body: Details about application compute resource requirements, associated artefacts, QoS profile and regions where application shall be made available etc.
		:type body: dict \| bytes
		:param federation_context_id:
		:type federation_context_id: dict \| bytes

		:rtype: None
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		try:
		# Find federation at originating OP
		originating_op_objects = OriginatingOperatorPlatformOriginatingOP.objects(id=federation_context_id)
		if not originating_op_objects:
		raise APIError(404, "Federation not found")
		originating_op_instance = originating_op_objects[0]

		# Find application onboarding at partner
		response_get = fm_client.get_profile(originating_op_instance.partner_federation_id, body.app_id, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_get and "status_code" in response_get:
		# If it's a 404, that means application doesn't exist at partner, which is what we want for onboarding
		if response_get["status_code"] != 404:
		status_code = response_get["status_code"]
		error_message = response_get["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		# 404 means application doesn't exist at partner, continue with onboarding logic
		if find_application_onboarding_at_originating_op(federation_context_id, body.app_id):
		raise APIError(409, "Application onboarding exist in originating operator but not exist at partner operator")
		else:
		# Create application profile at partner
		response_data = fm_client.create_profile(originating_op_instance.partner_federation_id, body, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_data and "status_code" in response_data:
		status_code = response_data["status_code"]
		error_message = response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "accepted" in response_data:
		# Create application onboarding in originating
		create_application_onboarding_originating_op(federation_context_id,
		originating_op_instance.partner_federation_id, body)
		return response_data, 202
		else:
		raise APIError(422, f"Unexpected response from partner API: {response_data}")
		elif "appId" in response_get:
		if find_application_onboarding_at_originating_op(federation_context_id, body.app_id):
		raise APIError(409, "Application onboarding already exists")
		else:
		raise APIError(409, "Application onboarding exist at partner operator but not exist in originating operator")
		else:
		if find_application_onboarding_at_originating_op(federation_context_id, body.app_id):
		raise APIError(409, "Application onboarding exist in originating operator but not exist at partner operator")
		else:
		# Create application profile at partner
		response_data = fm_client.create_profile(originating_op_instance.partner_federation_id, body, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_data and "status_code" in response_data:
		status_code = response_data["status_code"]
		error_message = response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "accepted" in response_data:
		# Create application onboarding in originating
		create_application_onboarding_originating_op(federation_context_id,
		originating_op_instance.partner_federation_id, body)
		return response_data, 202
		else:
		raise APIError(422, f"Unexpected response from partner API: {response_data}")
		except ValidationError:
		raise APIError(400, f"Invalid federation context ID format: {federation_context_id}")
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while onboarding application. Reason: {error}")


		def update_application(body, federation_context_id, app_id, bearer_token, partner_api_root): # noqa: E501
		"""Updates partner OP about changes in application compute resource requirements, QOS Profile, associated descriptor or change in associated components

		# noqa: E501

		:param body: Details about application compute resource requirements, associated artefact and QOS profile that needs to be updated.
		:type body: dict \| bytes
		:param federation_context_id:
		:type federation_context_id: dict \| bytes
		:param app_id:
		:type app_id: dict \| bytes

		:rtype: None
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		try:
		# Find federation at originating OP
		originating_op_objects = OriginatingOperatorPlatformOriginatingOP.objects(id=federation_context_id)
		if not originating_op_objects:
		raise APIError(404, "Federation not found")
		originating_op_instance = originating_op_objects[0]

		# Find application onboarding at partner
		response_get = fm_client.get_profile(originating_op_instance.partner_federation_id, app_id, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_get and "status_code" in response_get:
		status_code = response_get["status_code"]
		error_message = response_get["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "appId" in response_get:
		originating_ao_objects = find_application_onboarding_at_originating_op(federation_context_id, app_id)
		if originating_ao_objects:
		# Update application onboarding at partner
		response_data = fm_client.update_profile(originating_op_instance.partner_federation_id, app_id, body,
		bearer_token, partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in response_data and "status_code" in response_data:
		status_code = response_data["status_code"]
		error_message = response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "accepted" in response_data:
		originating_ao_instance = originating_ao_objects[0]
		update_application_onboarding_originating_op(body, federation_context_id,
		originating_op_instance.partner_federation_id, app_id,
		originating_ao_instance)
		return response_data, 202
		else:
		raise APIError(422, f"Unexpected response from partner API: {response_data}")
		else:
		raise APIError(409, "Application onboarding exist at partner operator but not exist in originating operator")
		else:
		if find_application_onboarding_at_originating_op(federation_context_id, app_id):
		raise APIError(409, "Application onboarding exist in originating operator but not exist at partner operator")
		else:
		raise APIError(404, "Application onboarding not found")
		except ValidationError:
		raise APIError(400, f"Invalid federation context ID or app ID format: {federation_context_id}, {app_id}")
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while updating application. Reason: {error}")


		def view_application(federation_context_id, app_id, bearer_token, partner_api_root): # noqa: E501
		"""Retrieves application details from partner OP

		# noqa: E501

		:param federation_context_id:
		:type federation_context_id: dict \| bytes
		:param app_id:
		:type app_id: dict \| bytes

		:rtype: InlineResponse2007
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		try:
		# Find federation at originating OP
		originating_op_objects = OriginatingOperatorPlatformOriginatingOP.objects(id=federation_context_id)
		if not originating_op_objects:
		raise APIError(404, "Federation not found")
		originating_op_instance = originating_op_objects[0]

		application_response_data = fm_client.get_profile(originating_op_instance.partner_federation_id, app_id,
		bearer_token, partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in application_response_data and "status_code" in application_response_data:
		status_code = application_response_data["status_code"]
		error_message = application_response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "appId" in application_response_data:
		if find_application_onboarding_at_originating_op(federation_context_id, app_id):
		return application_response_data, 200
		else:
		raise APIError(409, "Application Onboarding exist at partner operator but not in originating operator")
		else:
		if find_application_onboarding_at_originating_op(federation_context_id, app_id):
		raise APIError(409, "Application Onboarding exist in originating operator but not at partner operator")
		else:
		raise APIError(404, "Application Onboarding not found")
		except ValidationError:
		raise APIError(400, f"Invalid federation context ID or app ID format: {federation_context_id}, {app_id}")
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while viewing application. Reason: {error}")


		def find_application_onboarding_at_originating_op(federation_id, app_id):
		originating_ao_objects = OriginatingApplicationOnboardingManagementOriginatingOP.objects(
		orig_ao_federation_context_id=federation_id,
		orig_ao_app_id=app_id)

		return originating_ao_objects


		def delete_application_onboarding_originating_op(originating_ao_objects):
		# Delete all the onboarding updates related to onboarding application
		obj_onboarding = originating_ao_objects.get()
		id_onboarding = obj_onboarding.id
		originating_ao_update_objects = OriginatingApplicationOnboardingManagementUpdateOriginatingOP.objects()
		for o in originating_ao_update_objects:
		try:
		if o.federation_context_app_id.pk == id_onboarding:
		o.delete()
		except Exception as error:
		print(f"Unable to delete onboarding updates. Error: {error}")

		# Delete Application Onboarding
		originating_ao_objects.delete()


		def create_application_onboarding_originating_op(federation_id, partner_federation_id, body):
		onboarding_data = {
		"orig_ao_federation_context_id": federation_id,
		"orig_ao_app_id": body.app_id,
		"orig_ao_app_provider_id": body.app_provider_id,
		"orig_ao_app_deployment_zones": body.app_deployment_zones,
		"orig_ao_app_meta_data_app_name": body.app_meta_data.app_name,
		"orig_ao_app_meta_data_version": body.app_meta_data.version,
		"orig_ao_app_meta_data_app_description": body.app_meta_data.app_description,
		"orig_ao_app_meta_data_mobility_support": body.app_meta_data.mobility_support,
		"orig_ao_app_meta_data_access_token": body.app_meta_data.access_token,
		"orig_ao_app_meta_data_category": body.app_meta_data.category,
		"orig_ao_app_qos_profile_latency_constraints": body.app_qo_s_profile.latency_constraints,
		"orig_ao_app_qos_profile_bandwidth_required": body.app_qo_s_profile.bandwidth_required,
		"orig_ao_app_qos_profile_multi_user_clients": body.app_qo_s_profile.multi_user_clients,
		"orig_ao_app_qos_profile_no_of_users_per_app_inst": body.app_qo_s_profile.no_of_users_per_app_inst,
		"orig_ao_app_qos_profile_app_provisioning": body.app_qo_s_profile.app_provisioning,
		"orig_ao_app_component_specs": json.dumps(body.app_component_specs),
		"orig_ao_app_status_callback_link": body.app_status_callback_link,
		"partner_federation_id": partner_federation_id
		}

		# Create a new MongoEngine document and save it to MongoDB
		new_onboarding = OriginatingApplicationOnboardingManagementOriginatingOP(**onboarding_data)
		new_onboarding.save()


		def update_application_onboarding_originating_op(body, federation_context_id, partner_federation_id, app_id, originating_ao_instance):
		app_update_component_specs_list = []
		for acs in body.app_component_specs:
		data_acs = {
		"artefactId": acs.get("artefactId")
		}
		if acs.get("serviceNameNB"):
		data_acs["serviceNameNB"] = acs.get("serviceNameNB")
		if acs.get("serviceNameEW"):
		data_acs["serviceNameEW"] = acs.get("serviceNameEW")
		if acs.get("componentName"):
		data_acs["componentName"] = acs.get("componentName")
		app_update_component_specs_list.append(data_acs)

		onboarding_update_data = {
		"app_component_specs": json.dumps(app_update_component_specs_list),
		"federation_context_app_id": originating_ao_instance,
		"partner_federation_id": partner_federation_id
		}

		if body.app_upd_qo_s_profile:
		if body.app_upd_qo_s_profile.latency_constraints:
		originating_ao_instance.update(
		orig_ao_app_qos_profile_latency_constraints=body.app_upd_qo_s_profile.latency_constraints)
		if body.app_upd_qo_s_profile.bandwidth_required:
		originating_ao_instance.update(
		orig_ao_app_qos_profile_bandwidth_required=body.app_upd_qo_s_profile.bandwidth_required)
		if body.app_upd_qo_s_profile.multi_user_clients:
		originating_ao_instance.update(
		orig_ao_app_qos_profile_multi_user_clients=body.app_upd_qo_s_profile.multi_user_clients)
		if body.app_upd_qo_s_profile.no_of_users_per_app_inst:
		originating_ao_instance.update(
		orig_ao_app_qos_profile_no_of_users_per_app_inst=body.app_upd_qo_s_profile.no_of_users_per_app_inst)
		if body.app_upd_qo_s_profile.app_provisioning:
		originating_ao_instance.update(
		orig_ao_app_qos_profile_app_provisioning=body.app_upd_qo_s_profile.app_provisioning)
		if body.app_upd_qo_s_profile.mobility_support:
		originating_ao_instance.update(
		orig_ao_app_meta_data_mobility_support=body.app_upd_qo_s_profile.mobility_support)
		originating_ao_instance.update(orig_ao_app_component_specs=onboarding_update_data.get("app_component_specs"))

		# Create a new MongoEngine document and save it to MongoDB
		new_onboarding_update = OriginatingApplicationOnboardingManagementUpdateOriginatingOP(**onboarding_update_data)
		new_onboarding_update.save()

src/adapters/tf_adapter/application_onboarding_management.py

0 → 100644

+451 −0

File added.

Preview size limit exceeded, changes collapsed.

src/api/application_onboarding_management.py

0 → 100644

+139 −0

File added.

Preview size limit exceeded, changes collapsed.

src/controllers/application_onboarding_management_controller.py

deleted100644 → 0

+0 −790

File deleted.

Preview size limit exceeded, changes collapsed.