Adapts federation management API (e35a2f10) · Commits · OOP / code / federation-manager

src/adapters/fm_adapter/federation_management.py

0 → 100644

+333 −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. #
		# -------------------------------------------------------------------------- #
		from configparser import ConfigParser
		from mongoengine.errors import ValidationError


		from models.federation_context_id import FederationContextId # noqa: E501
		from models.federation_response_data import FederationResponseData # noqa: E501
		from models.inline_response2001 import InlineResponse2001 # noqa: E501
		from models.mongo_document import OriginatingOperatorPlatformOriginatingOP
		from models.mongo_document import OriginatingOperatorPlatformUpdateOriginatingOP
		from clients import fed_manager as fm_client
		from adapters.error import APIError


		CONFIG = ConfigParser()
		CONFIG.read("conf/config.cfg")
		partnerOPFederationId = CONFIG.get("op_data", "partnerOPFederationId")
		partnerOPCountryCode = CONFIG.get("op_data", "partnerOPCountryCode")
		partnerOPMobileNetworkCode_MCC = CONFIG.get("op_data", "partnerOPMobileNetworkCode_MCC")
		partnerOPMobileNetworkCode_MNC = CONFIG.get("op_data", "partnerOPMobileNetworkCode_MNC")
		partnerOPFixedNetworkCode = CONFIG.get("op_data", "partnerOPFixedNetworkCode")
		partnerOPPlatformCaps = CONFIG.get("op_data", "platformCaps")


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


		def create_federation(body, bearer_token, partner_api_root): # noqa: E501
		"""Creates one direction federation with partner operator platform.

		# noqa: E501

		:param body:
		:type body: dict \| bytes

		:rtype: FederationResponseData
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		# Forward request to partner OP
		try:
		federation_response_data = fm_client.create_federation(body, bearer_token, partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in federation_response_data and "status_code" in federation_response_data:
		status_code = federation_response_data["status_code"]
		error_message = federation_response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif federation_response_data.get("federationContextId"):
		federation_response_data = create_federation_at_originating_op(body, bearer_token,
		federation_response_data)
		federation_response_data = FederationResponseData.from_dict(federation_response_data)
		return federation_response_data, 200
		else:
		return federation_response_data, 422
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while creating Federation. Reason: {error}")


		def get_federation_details(federation_context_id, bearer_token, partner_api_root): # noqa: E501
		"""Retrieves details about the federation context with the partner OP. The response shall provide info about the zones offered by the partner, partner OP network codes, information about edge discovery and LCM service etc.

		# noqa: E501

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

		:rtype: InlineResponse2001
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		# Forward request to partner OP
		try:
		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]

		federation_response_data = fm_client.get_federation(originating_op_instance.partner_federation_id, bearer_token,
		partner_api_root)

		# Check if the response contains an error from the partner API
		if "error" in federation_response_data and "status_code" in federation_response_data:
		status_code = federation_response_data["status_code"]
		error_message = federation_response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "edgeDiscoveryServiceEndPoint" in federation_response_data:
		federation_response_data = InlineResponse2001.from_dict(federation_response_data)
		return federation_response_data, 200
		else:
		raise APIError(422, f"Conflict between originating operator and partner operator. Reason: {federation_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 getting Federation. Reason: {error}")


		def update_federation(federation_context_id, body, bearer_token, partner_api_root): # noqa: E501
		"""API used by the Originating OP towards the partner OP, to update the parameters associated to the existing federation

		# noqa: E501

		:param body: Details about changes origination OP wished to apply
		:type body: dict \| bytes
		:param federation_context_id:
		:type federation_context_id: dict \| bytes

		:rtype: InlineResponse2001
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		# Forward request to partner OP
		try:
		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]

		federation_response_data = fm_client.update_federation(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 federation_response_data and "status_code" in federation_response_data:
		status_code = federation_response_data["status_code"]
		error_message = federation_response_data["error"]
		raise APIError(status_code, f"Partner API error: {error_message}")
		elif "edgeDiscoveryServiceEndPoint" in federation_response_data:
		federation_response_data = update_federation_at_originating_op(originating_op_instance, body,
		federation_response_data)
		federation_response_data = InlineResponse2001.from_dict(federation_response_data)
		return federation_response_data, 200
		else:
		raise APIError(422, f"Conflict between originating operator and partner operator. Reason: {federation_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 updating Federation. Reason: {error}")


		def delete_federation_details(federation_context_id, bearer_token, partner_api_root): # noqa: E501
		"""Remove existing federation with the partner OP

		# noqa: E501

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

		:rtype: None
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		# Forward request to partner OP
		try:
		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]

		response_data = fm_client.delete_federation(originating_op_instance.partner_federation_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 "removed" in response_data:
		delete_federation_originating_op(originating_op_instance)
		return response_data, 200
		else:
		raise APIError(422, f"Conflict between originating operator and partner operator. Reason: {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 deleting Federation. Reason: {error}")


		def get_federation_context_id(bearer_token, partner_api_root): # noqa: E501
		"""Retrieves the existing federationContextId with partner operator platform.

		# noqa: E501

		:rtype: InlineResponse2002
		"""

		# Verify partner API root is provided
		verify_required_header(partner_api_root)

		try:
		originating_op_objects = OriginatingOperatorPlatformOriginatingOP.objects(partner_bearer_token=bearer_token)
		if not originating_op_objects:
		raise APIError(404, "Federation not found")
		originating_op_instance = originating_op_objects[0]
		federation_context_id_data = {
		"federationContextId": str(originating_op_instance.id)
		}
		federation_context_id = FederationContextId.from_dict(federation_context_id_data)
		return federation_context_id
		except APIError:
		raise # Re-raise APIError as-is
		except Exception as error:
		raise APIError(500, f"Error while retrieving federation context ID. Reason: {error}")


		def create_federation_at_originating_op(body, bearer_token, federation_response_data):
		federation_data = {
		"orig_op_federation_id": body.orig_op_federation_id,
		"orig_op_country_code": body.orig_op_country_code,
		"orig_op_mobile_network_codes_mcc": body.orig_op_mobile_network_codes.mcc,
		"orig_op_mobile_network_codes_mncs": body.orig_op_mobile_network_codes.mncs,
		"orig_op_fixed_network_codes": body.orig_op_fixed_network_codes,
		"initial_date": body.initial_date,
		"partner_status_link": body.partner_status_link,
		"partner_callback_credentials_token_url": body.partner_callback_credentials.token_url,
		"partner_callback_credentials_client_id": body.partner_callback_credentials.client_id,
		"partner_callback_credentials_client_secret": body.partner_callback_credentials.client_secret,
		"partner_bearer_token": bearer_token,
		"partner_federation_id": federation_response_data.get("federationContextId")
		}

		# Create a new MongoEngine document and save it to MongoDB
		new_federation = OriginatingOperatorPlatformOriginatingOP(**federation_data)
		new_federation.save()

		response_data = federation_response_data
		response_data["federationContextId"] = str(new_federation.id)

		return response_data


		def update_federation_at_originating_op(originating_op_instance, body, federation_response_data):
		partner_update_data = {
		"object_type": body.object_type,
		"operation_type": body.operation_type,
		"modification_date": body.modification_date,
		"federation_context_id": str(originating_op_instance.id),
		"partner_federation_id": originating_op_instance.partner_federation_id
		}

		if body.object_type == "MOBILE_NETWORK_CODES":
		mncs = originating_op_instance.orig_op_mobile_network_codes_mncs
		if body.operation_type == "ADD_CODES":
		partner_update_data["add_mobile_network_ids_mcc"] = body.add_mobile_network_ids.mcc
		partner_update_data["add_mobile_network_ids_mncs"] = body.add_mobile_network_ids.mncs
		mncs.extend(body.add_mobile_network_ids.mncs)
		originating_op_instance.update(set__orig_op_mobile_network_codes_mncs=mncs)
		elif body.operation_type == "REMOVE_CODES":
		partner_update_data["remove_mobile_network_ids_mcc"] = body.remove_mobile_network_ids.mcc
		partner_update_data["remove_mobile_network_ids_mncs"] = body.remove_mobile_network_ids.mncs
		mncs = [code for code in mncs if code not in body.remove_mobile_network_ids.mncs]
		originating_op_instance.update(set__orig_op_mobile_network_codes_mncs=mncs)
		elif body.operation_type == "UPDATE_CODES":
		partner_update_data["add_mobile_network_ids_mcc"] = body.add_mobile_network_ids.mcc
		partner_update_data["add_mobile_network_ids_mncs"] = body.add_mobile_network_ids.mncs
		partner_update_data["remove_mobile_network_ids_mcc"] = body.remove_mobile_network_ids.mcc
		partner_update_data["remove_mobile_network_ids_mncs"] = body.remove_mobile_network_ids.mncs
		originating_op_instance.update(set__orig_op_mobile_network_codes_mcc=body.add_mobile_network_ids.mcc)
		mncs = body.add_mobile_network_ids.mncs
		if body.remove_mobile_network_ids.mcc == body.add_mobile_network_ids.mcc:
		mncs = [code for code in body.add_mobile_network_ids.mncs if code not in body.remove_mobile_network_ids.mncs]
		originating_op_instance.update(set__orig_op_mobile_network_codes_mncs=mncs)
		elif body.object_type == "FIXED_NETWORK_CODES":
		original_fixed_codes = originating_op_instance.orig_op_fixed_network_codes
		if body.operation_type == "ADD_CODES":
		partner_update_data["add_fixed_network_ids"] = body.add_fixed_network_ids
		fixed_codes = original_fixed_codes
		fixed_codes.extend(body.add_fixed_network_ids)
		originating_op_instance.update(set__orig_op_fixed_network_codes=fixed_codes)
		if body.operation_type == "REMOVE_CODES":
		partner_update_data["remove_fixed_network_ids"] = body.remove_fixed_network_ids
		fixed_codes = [code for code in original_fixed_codes if code not in body.remove_fixed_network_ids]
		originating_op_instance.update(set__orig_op_fixed_network_codes=fixed_codes)
		elif body.operation_type == "UPDATE_CODES":
		partner_update_data["add_fixed_network_ids"] = body.add_fixed_network_ids
		partner_update_data["remove_fixed_network_ids"] = body.remove_fixed_network_ids
		fixed_codes = [code for code in body.add_fixed_network_ids if code not in body.remove_fixed_network_ids]
		originating_op_instance.update(set__orig_op_fixed_network_codes=fixed_codes)

		# Create a new MongoEngine document and save it to MongoDB
		new_federation = OriginatingOperatorPlatformUpdateOriginatingOP(**partner_update_data)
		new_federation.save()

		response_data = federation_response_data
		response_data["federationContextId"] = str(originating_op_instance.id)

		return response_data


		def delete_federation_originating_op(originating_op_instance):
		# Delete all the federation updates related to federation
		id_federation = originating_op_instance.id
		originating_op_instance_update_objects = OriginatingOperatorPlatformUpdateOriginatingOP.objects()
		for o in originating_op_instance_update_objects:
		try:
		if o.federation_context_id.pk == id_federation:
		o.delete()
		except Exception:
		pass

		# Delete Federation
		originating_op_instance.delete()

src/adapters/tf_adapter/federation_management.py

0 → 100644

+400 −0

File added.

Preview size limit exceeded, changes collapsed.

src/api/federation_management.py

0 → 100644

+155 −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 connexion
		from configparser import ConfigParser
		from flask import abort

		from adapters.error import APIError
		import util

		from models.federation_request_data import FederationRequestData # noqa: E501
		from models.federation_context_id_partner_body import FederationContextIdPartnerBody # noqa: E501
		from adapters.injector import resolve_adapter

		CONFIG = ConfigParser()
		CONFIG.read("conf/config.cfg")
		partnerOPFederationId = CONFIG.get("op_data", "partnerOPFederationId")
		partnerOPCountryCode = CONFIG.get("op_data", "partnerOPCountryCode")
		partnerOPMobileNetworkCode_MCC = CONFIG.get("op_data", "partnerOPMobileNetworkCode_MCC")
		partnerOPMobileNetworkCode_MNC = CONFIG.get("op_data", "partnerOPMobileNetworkCode_MNC")
		partnerOPFixedNetworkCode = CONFIG.get("op_data", "partnerOPFixedNetworkCode")
		platformCaps = CONFIG.get("op_data", "platformCaps")


		def create_federation(body): # noqa: E501
		"""Creates one direction federation with partner operator platform.

		# noqa: E501

		:param body:
		:type body: dict \| bytes

		:rtype: FederationResponseData
		"""
		# Extract the token and the headers
		bearer_token = util.get_token_from_request(connexion)
		headers = dict(connexion.request.headers)
		partner_api_root = headers.get("X-Partner-Api-Root")

		try:
		body = FederationRequestData.from_dict(connexion.request.get_json()) # noqa: E501
		except Exception as error:
		abort(422, f"Federation Validation Error. Message: {error}")
		try:
		adapter = resolve_adapter(headers)
		return adapter.federation_management.create_federation(body, bearer_token, partner_api_root)
		except APIError as error:
		abort(error.status_code, error.detail_error)


		def get_federation_details(federation_context_id): # noqa: E501
		"""Retrieves details about the federation context with the partner OP. The response shall provide info about the zones offered by the partner, partner OP network codes, information about edge discovery and LCM service etc.

		# noqa: E501

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

		:rtype: InlineResponse2001
		"""
		# Extract the token and the headers
		bearer_token = util.get_token_from_request(connexion)
		headers = dict(connexion.request.headers)
		partner_api_root = headers.get("X-Partner-Api-Root")

		try:
		adapter = resolve_adapter(headers)
		return adapter.federation_management.get_federation_details(federation_context_id, bearer_token,
		partner_api_root)
		except APIError as error:
		abort(error.status_code, error.detail_error)


		def update_federation(federation_context_id, body): # noqa: E501
		"""API used by the Originating OP towards the partner OP, to update the parameters associated to the existing federation

		# noqa: E501

		:param body: Details about changes origination OP wished to apply
		:type body: dict \| bytes
		:param federation_context_id:
		:type federation_context_id: dict \| bytes

		:rtype: InlineResponse2001
		"""
		# Extract the token and the headers
		bearer_token = util.get_token_from_request(connexion)
		headers = dict(connexion.request.headers)
		partner_api_root = headers.get("X-Partner-Api-Root")

		try:
		body = FederationContextIdPartnerBody.from_dict(body) # noqa: E501
		except Exception as error:
		raise APIError(422, f"Federation Validation Error. Message: {error}")

		try:
		adapter = resolve_adapter(headers)
		return adapter.federation_management.update_federation(federation_context_id, body, bearer_token,
		partner_api_root)
		except APIError as error:
		abort(error.status_code, error.detail_error)


		def delete_federation_details(federation_context_id): # noqa: E501
		"""Remove existing federation with the partner OP

		# noqa: E501

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

		:rtype: None
		"""
		# Extract the token and the headers
		bearer_token = util.get_token_from_request(connexion)
		headers = dict(connexion.request.headers)
		partner_api_root = headers.get("X-Partner-Api-Root")

		try:
		adapter = resolve_adapter(headers)
		return adapter.federation_management.delete_federation_details(federation_context_id, bearer_token,
		partner_api_root)
		except APIError as error:
		abort(error.status_code, error.detail_error)


		def get_federation_context_id(): # noqa: E501
		"""Retrieves the existing federationContextId with partner operator platform.

		# noqa: E501

		:rtype: InlineResponse2002
		"""
		# Extract the token and the headers
		bearer_token = util.get_token_from_request(connexion)
		headers = dict(connexion.request.headers)
		partner_api_root = headers.get("X-Partner-Api-Root")

		try:
		adapter = resolve_adapter(headers)
		return adapter.federation_management.get_federation_context_id(bearer_token, partner_api_root)
		except APIError as error:
		abort(error.status_code, error.detail_error)

src/controllers/federation_management_controller.py

deleted100644 → 0

+0 −587

File deleted.

Preview size limit exceeded, changes collapsed.