update federation manager and service (115dc2be) · Commits · OOP / code / open-exposure-gateway

edge_cloud_management_api/controllers/federation_manager_controller.py

+41 −0

Changes for edge_cloud_management_api/controllers/federation_manager_controller.py: 41 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -51,3 +51,44 @@ def get_federation_context_ids():
		return jsonify(result), 200 if "error" not in result else 502
		except Exception as e:
		return jsonify({"error": str(e)}), 400




		def onboard_application_to_partner(federationContextId):
		"""
		POST /{federationContextId}/application/onboarding
		Forwards the onboarding request to the Federation Manager.
		"""
		try:
		body = request.get_json()
		result = federation_client.onboard_application(federationContextId, body)
		return jsonify(result), 202 if "error" not in result else 502
		except Exception as e:
		return jsonify({"error": str(e)}), 400

		def get_onboarded_app(federationContextId, appId):
		"""
		GET /{federationContextId}/application/onboarding/app/{appId}
		Retrieves onboarding info of a federated app from a partner OP.
		"""
		try:
		result = federation_client.get_onboarded_app(federationContextId, appId)
		if "error" in result:
		return jsonify(result), result.get("status_code", 502)
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 500

		def delete_onboarded_app(federationContextId, appId):
		"""
		DELETE /{federationContextId}/application/onboarding/app/{appId}
		Deboards the application and deletes it from the partner OP.
		"""
		try:
		result = federation_client.delete_onboarded_app(federationContextId, appId)
		if "error" in result:
		return jsonify(result), result.get("status_code", 502)
		return jsonify({"message": f"App {appId} successfully deleted from partner"}), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 500

edge_cloud_management_api/models/federation_manager_models.py

+1 −1

Changes for edge_cloud_management_api/models/federation_manager_models.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		from pydantic import BaseModel
		from pydantic import BaseModel, Field
		from typing import List, Optional

edge_cloud_management_api/services/federation_services.py

+106 −0

Changes for edge_cloud_management_api/services/federation_services.py: 106 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -90,6 +90,65 @@ class FederationManagerClient:
		logger.error(f"GET /fed-context-id unexpected error: {e}")
		return {"error": str(e)}

		def onboard_application(self, federation_context_id: str, body: dict):
		url = f"{self.base_url}/{federation_context_id}/application/onboarding"
		try:
		response = requests.post(url, headers=self._get_headers(), json=body, timeout=10)
		response.raise_for_status()
		return response.json()
		except Timeout:
		logger.error("POST /application/onboarding timed out")
		return {"error": "Request timed out"}
		except ConnectionError:
		logger.error("POST /application/onboarding connection error")
		return {"error": "Connection error"}
		except requests.exceptions.HTTPError as http_err:
		logger.error(f"POST /application/onboarding HTTP error: {http_err}")
		return {"error": str(http_err), "status_code": response.status_code}
		except Exception as e:
		logger.error(f"POST /application/onboarding unexpected error: {e}")
		return {"error": str(e)}


		def get_onboarded_app(self, federation_context_id: str, app_id: str):
		url = f"{self.base_url}/{federation_context_id}/application/onboarding/app/{app_id}"
		try:
		response = requests.get(url, headers=self._get_headers(), timeout=10)
		response.raise_for_status()
		return response.json()
		except Timeout:
		logger.error("GET onboarded app timed out")
		return {"error": "Request timed out", "status_code": 408}
		except ConnectionError:
		logger.error("GET onboarded app connection error")
		return {"error": "Connection error", "status_code": 503}
		except requests.exceptions.HTTPError as http_err:
		logger.error(f"GET onboarded app HTTP error: {http_err}")
		return {"error": str(http_err), "status_code": response.status_code}
		except Exception as e:
		logger.error(f"GET onboarded app unexpected error: {e}")
		return {"error": str(e), "status_code": 500}

		def delete_onboarded_app(self, federation_context_id: str, app_id: str):
		url = f"{self.base_url}/{federation_context_id}/application/onboarding/app/{app_id}"
		try:
		response = requests.delete(url, headers=self._get_headers(), timeout=10)
		response.raise_for_status()
		return {"message": "Deleted successfully", "status_code": response.status_code}
		except Timeout:
		logger.error("DELETE onboarding app timed out")
		return {"error": "Request timed out", "status_code": 504}
		except ConnectionError:
		logger.error("DELETE onboarding app connection error")
		return {"error": "Connection error", "status_code": 503}
		except requests.exceptions.HTTPError as http_err:
		logger.error(f"DELETE onboarding app HTTP error: {http_err}")
		return {"error": str(http_err), "status_code": response.status_code}
		except Exception as e:
		logger.error(f"DELETE onboarding app unexpected error: {e}")
		return {"error": str(e), "status_code": 500}



		class FederationManagerClientFactory:
		def __init__(self):
		@@ -99,6 +158,53 @@ class FederationManagerClientFactory:
		base_url = base_url or self.default_base_url
		return FederationManagerClient(base_url=base_url)

		def onboard_application_to_partners(app_id, zones):
		SOURCE_OP = os.getenv("SOURCE_OP_ID")
		federation_context_id = os.getenv("FEDERATION_CONTEXT_ID")
		callback_url = os.getenv("STATUS_CALLBACK_URL", "http://your-callback/api/status")

		partner_zones = [zone for zone in zones if zone.get("edgeCloudProvider") != SOURCE_OP]
		if not partner_zones:
		return {"message": "No partner zones to onboard"}, 200

		srm_client = PiEdgeAPIClientFactory().create_pi_edge_api_client()
		app_data = srm_client.get_app(app_id)

		if not app_data or "error" in app_data:
		return {"error": "Failed to retrieve application metadata from SRM"}, 500

		app_manifest = app_data.get("appManifest", {})
		onboarding_payload = {
		"appId": app_id,
		"appManifest": app_manifest,
		"zones": partner_zones,
		"appStatusCallbackLink": callback_url
		}

		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()

		results = []
		for zone in partner_zones:
		partner_op = zone["edgeCloudProvider"]
		try:
		response = federation_client.onboard_application(federation_context_id, onboarding_payload)
		results.append({
		"zoneId": zone.get("edgeCloudZoneId"),
		"provider": partner_op,
		"status": response.get("status", "success"),
		"detail": response
		})
		except Exception as e:
		results.append({
		"zoneId": zone.get("edgeCloudZoneId"),
		"provider": partner_op,
		"status": "error",
		"detail": str(e)
		})

		return {"onboardingResults": results}, 202


		if __name__ == "__main__":
		factory = FederationManagerClientFactory()