Updates client to interact via EWBI with other partners OP (33139876) · Commits · OOP / code / federation-manager

src/clients/fed_manager.py

+297 −80

Original line number	Diff line number	Diff line
		@@ -13,190 +13,407 @@
		# See the License for the specific language governing permissions and #
		# limitations under the License. #
		# -------------------------------------------------------------------------- #

		import requests
		from configparser import ConfigParser
		from urllib.parse import urlparse


		def check_url(api_root, path):

		url = f"{api_root}/operatorplatform/federation/v1/{path}"
		parsed_url = urlparse(url)
		try:
		if parsed_url.scheme and parsed_url.netloc:
		requests.head(url, timeout=5)
		return url
		except requests.RequestException:
		return None


		CONFIG = ConfigParser()
		CONFIG.read("conf/config.cfg")
		SERVER_PARTNER = CONFIG.get("partner_op", "partner_op_server")
		PORT_PARTNER = CONFIG.get("partner_op", "partner_op_port")
		HOST_PARTNER = CONFIG.get("partner_op", "partner_op_host")
		URL_PARTNER = f"{HOST_PARTNER}:{PORT_PARTNER}{SERVER_PARTNER}"
		def create_federation(body, token, api_root):

		def create_federation(token, body):
		headers = {
		"Content-Type": "application/json",
		"Accept": "application/json",
		"Authorization": f"Bearer {token}"
		}

		url = f"http://{URL_PARTNER}/partner"
		partner_op_url = check_url(api_root, "partner")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		json_payload = body.to_gsma_input()

		response = requests.post(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def get_federation(federation_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/partner"
		def get_federation(federation_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/partner")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		response = requests.get(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def update_federation(token, federation_id, body):

		url = f"http://{URL_PARTNER}/{federation_id}/partner"
		def update_federation(federation_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/partner")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		response = requests.patch(partner_op_url, headers=headers, json=json_payload)

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.patch(url, headers=headers, json=body)
		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def delete_federation(federation_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/partner"
		def delete_federation(federation_id, token, api_root):

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/partner")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.delete(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def create_availability_zones(federation_context_id, body, token):

		url = f"http://{URL_PARTNER}/{federation_context_id}/zones"
		def get_federation_context_id(token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		partner_op_url = check_url(api_root, "fed-context-id")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.get(partner_op_url, headers=headers)

		return response.json()


		def create_availability_zones(federation_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/zones")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		response = requests.post(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def get_availability_zones(federation_id, zone_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/zones/{zone_id}"
		def get_availability_zones(federation_id, zone_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/zones/{zone_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.get(partner_op_url, headers=headers)

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Zone not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def delete_availability_zones(federation_id, zone_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/zones/{zone_id}"
		def delete_availability_zones(federation_id, zone_id, token, api_root):

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/zones/{zone_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.delete(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Zone not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def create_artefact(federation_context_id, body, token):

		url = f"http://{URL_PARTNER}/{federation_context_id}/artefact"
		def create_artefact(federation_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		partner_op_url = check_url(api_root, f"{federation_id}/artefact")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		response = requests.post(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response


		def get_artefact(federation_id, artefact_id, token):
		def get_artefact(federation_id, artefact_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/artefact/{artefact_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		url = f"http://{URL_PARTNER}/{federation_id}/artefact/{artefact_id}"
		response = requests.get(partner_op_url, headers=headers)

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Artefact not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def delete_artefact(federation_id, artefact_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/artefact/{artefact_id}"
		def delete_artefact(federation_id, artefact_id, token, api_root):

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/artefact/{artefact_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.delete(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Artefact not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def create_profile(federation_context_id, body, token):

		url = f"http://{URL_PARTNER}/{federation_context_id}/application/onboarding"
		def create_profile(federation_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		partner_op_url = check_url(api_root, f"{federation_id}/application/onboarding")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		response = requests.post(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def delete_profile(federation_id, app_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/application/onboarding/app/{app_id}"
		def delete_profile(federation_id, app_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/application/onboarding/app/{app_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		response = requests.delete(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def update_profile(token, federation_id, app_id, body):

		url = f"http://{URL_PARTNER}/{federation_id}/application/onboarding/app/{app_id}"
		def update_profile(federation_id, app_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/application/onboarding/app/{app_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.patch(url, headers=headers, json=body)
		response = requests.patch(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def get_profile(federation_id, app_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/application/onboarding/app/{app_id}"
		def get_profile(federation_id, app_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/application/onboarding/app/{app_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		response = requests.get(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def get_all_instances_deployment(federation_id, app_id, app_provider_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/application/lcm/app/{app_id}/appProvider/{app_provider_id}"
		def get_all_instances_deployment(federation_id, app_id, app_provider_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		partner_op_url = check_url(api_root,
		f"{federation_id}/application/lcm/app/{app_id}/appProvider/{app_provider_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.get(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application instances not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def get_instance_details_deployment(federation_id, app_id, app_instance_id, zone_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/application/lcm/app/{app_id}/instance/{app_instance_id}/zone/{zone_id}"
		def get_instance_details_deployment(federation_id, app_id, app_instance_id, zone_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root,
		f"{federation_id}/application/lcm/app/{app_id}/instance/{app_instance_id}/zone/{zone_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		response = requests.get(partner_op_url, headers=headers)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application instance not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

		def install_app_deployment(federation_context_id, body, token):

		url = f"http://{URL_PARTNER}/{federation_context_id}/application/lcm"
		def install_app_deployment(federation_id, body, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root, f"{federation_id}/application/lcm")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		json_payload = body.to_gsma_input()

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		response = requests.post(partner_op_url, headers=headers, json=json_payload)

		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Federation not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		data_response = response.json()

		return data_response

		def remove_app_deployment(federation_id, app_id, app_instance_id, zone_id, token):

		url = f"http://{URL_PARTNER}/{federation_id}/application/lcm/app/{app_id}/instance/{app_instance_id}/zone/{zone_id}"
		def remove_app_deployment(federation_id, app_id, app_instance_id, zone_id, token, api_root):

		headers = {"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"}

		partner_op_url = check_url(api_root,
		f"{federation_id}/application/lcm/app/{app_id}/instance/{app_instance_id}/zone/{zone_id}")
		if not partner_op_url:
		return {"error": f"Invalid URL: {partner_op_url}"}

		response = requests.delete(partner_op_url, headers=headers)

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		# Handle different HTTP status codes
		if response.status_code == 404:
		return {"error": "Application instance not found", "status_code": 404}
		elif response.status_code >= 400:
		return {"error": f"HTTP {response.status_code}: {response.text}", "status_code": response.status_code}

		return response.json()

src/conf/config.cfg.sample

+0 −8

Original line number	Diff line number	Diff line
		@@ -37,11 +37,3 @@ lcmServiceEndPoint_port = 8989
		lcmServiceEndPoint_fqdn =
		lcmServiceEndPoint_ipv4Addresses = 127.0.0.1
		lcmServiceEndPoint_ipv6Addresses =

		[partner_op]
		# Defines the role of the Federation Manager
		partner_op_host = 127.0.0.1
		partner_op_server = /operatorplatform/federation/v1
		partner_op_port = 8990
		#role = originating_op
		role = partner_op

src/util.py

+11 −3

Original line number	Diff line number	Diff line
		@@ -20,8 +20,6 @@ import six
		import type_util
		from enum import Enum

		_ROLE_ORIGINATING_OP = "originating_op"
		_ROLE_PARTNER_OP = "partner_op"

		def _deserialize(data, klass):
		"""Deserializes dict, list, str into an object.
		@@ -159,6 +157,7 @@ def _deserialize_dict(data, boxed_type):
		return {k: _deserialize(v, boxed_type)
		for k, v in six.iteritems(data)}


		def get_token_from_request(connection):
		bearer_token = ""

		@@ -172,6 +171,15 @@ def get_token_from_request(connection):

		return bearer_token


		def get_header_from_request(connection):

		# Access the Partner-API-Root header
		partner_api_root = connection.request.headers.get('X-Partner-API-Root')

		return partner_api_root


		class RepoType(Enum):
		PRIVATE = "PRIVATEREPO"
		PUBLIC = "PUBLICREPO"