Loading edge_cloud_management_api/controllers/app_controllers.py +7 −1 Original line number Diff line number Diff line from flask import jsonify, request from pydantic import ValidationError from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.storage_service import get_zone Loading Loading @@ -99,6 +100,11 @@ def create_app_instance(): if not app_id or not edge_zone_id : return jsonify({"error": "Missing required fields: appId, edgeCloudZoneId, or kubernetesCLusterRef"}), 400 local_zone = get_zone(edge_zone_id) if not local_zone: # TODO: apply federation logic return 'Zone belongs to federated partner OP. Switching to Federation Manager.' logger.info(f"Preparing to send deployment request to SRM for appId={app_id}") pi_edge_client_factory = PiEdgeAPIClientFactory() Loading edge_cloud_management_api/controllers/edge_cloud_controller.py +10 −3 Original line number Diff line number Diff line from flask import jsonify from pydantic import BaseModel, Field, ValidationError from typing import List from edge_cloud_management_api.configs.env_config import config from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.storage_service import insert_zones pi_edge_factory = PiEdgeAPIClientFactory() api_client = pi_edge_factory.create_pi_edge_api_client() zones = api_client.edge_cloud_zones() for zone in zones: zone['_id'] = zone.get('edgeCloudZoneId') insert_zones(zones) class EdgeCloudZone(BaseModel): edgeCloudZoneId: str = Field(..., description="Unique identifier of the Edge Cloud Zone") Loading edge_cloud_management_api/controllers/federation_manager_controller.py +39 −48 Original line number Diff line number Diff line from flask import request, jsonify, Response import json from flask import request, jsonify import logging import connexion from requests.exceptions import Timeout, ConnectionError from edge_cloud_management_api.managers.log_manager import logger import requests from edge_cloud_management_api.services.federation_services import FederationManagerClientFactory # Factory pattern Loading @@ -8,74 +12,61 @@ factory = FederationManagerClientFactory() federation_client = factory.create_federation_client() def create_federation(): """POST /partner - Create federation with partner OP.""" try: body = request.get_json() result = federation_client.post_partner(body) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 body = request.get_json() token = __get_token() response = federation_client.post_partner(body, token) return jsonify(response) def get_federation(federationContextId): """GET /{federationContextId}/partner - Get federation info.""" try: result = federation_client.get_partner(federationContextId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.get_partner(federationContextId, token) return jsonify(result) def delete_federation(federationContextId): """DELETE /{federationContextId}/partner - Delete federation.""" try: result = federation_client.delete_partner(federationContextId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.delete_partner(federationContextId, token) return jsonify(result) def get_federation_context_ids(): """GET /fed-context-id - Fetch federationContextId(s).""" try: result = federation_client.get_federation_context_ids() return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() response = federation_client.get_federation_context_ids(token) return jsonify(response) def onboard_application_to_partner(federationContextId): """POST /{federationContextId}/application/onboarding - Onboard app.""" try: body = request.get_json() result = federation_client.onboard_application(federationContextId, body) return jsonify(result), 202 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.onboard_application(federationContextId, body, token) return jsonify(result) def get_onboarded_app(federationContextId, appId): """GET /{federationContextId}/application/onboarding/app/{appId}""" try: result = federation_client.get_onboarded_app(federationContextId, appId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.get_onboarded_app(federationContextId, appId, token) return jsonify(result) def delete_onboarded_app(federationContextId, appId): """DELETE /{federationContextId}/application/onboarding/app/{appId}""" try: result = federation_client.delete_onboarded_app(federationContextId, appId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.delete_onboarded_app(federationContextId, appId, token) return jsonify(result) def __get_token(): bearer = connexion.request.headers['Authorization'] token = bearer.split()[1] return token edge_cloud_management_api/controllers/network_functions_controller.py +1 −1 Original line number Diff line number Diff line from flask import jsonify from pydantic import ValidationError #Field from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory def create_qod_session(body: dict): """ Loading edge_cloud_management_api/controllers/security_controller.py 0 → 100644 +18 −0 Original line number Diff line number Diff line import os from jose import JWTError, jwt from werkzeug.exceptions import Unauthorized def decode_token(token:str): JWT_ISSUER = os.environ.get('JWT_ISSUER') PUBLIC_KEY = os.environ.get('JWT_PUBLIC_KEY') try: return jwt.decode(token, key=PUBLIC_KEY, algorithms=["RS256"], issuer=JWT_ISSUER) except JWTError as e: raise Unauthorized from e def check_oAuth2ClientCredentials(token): return {'scopes': ['fed-mgmt'], 'uid': 'test_value'} def validate_scope_oAuth2ClientCredentials(required_scopes, token_scopes): return set(required_scopes).issubset(set(token_scopes)) Loading
edge_cloud_management_api/controllers/app_controllers.py +7 −1 Original line number Diff line number Diff line from flask import jsonify, request from pydantic import ValidationError from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.storage_service import get_zone Loading Loading @@ -99,6 +100,11 @@ def create_app_instance(): if not app_id or not edge_zone_id : return jsonify({"error": "Missing required fields: appId, edgeCloudZoneId, or kubernetesCLusterRef"}), 400 local_zone = get_zone(edge_zone_id) if not local_zone: # TODO: apply federation logic return 'Zone belongs to federated partner OP. Switching to Federation Manager.' logger.info(f"Preparing to send deployment request to SRM for appId={app_id}") pi_edge_client_factory = PiEdgeAPIClientFactory() Loading
edge_cloud_management_api/controllers/edge_cloud_controller.py +10 −3 Original line number Diff line number Diff line from flask import jsonify from pydantic import BaseModel, Field, ValidationError from typing import List from edge_cloud_management_api.configs.env_config import config from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.storage_service import insert_zones pi_edge_factory = PiEdgeAPIClientFactory() api_client = pi_edge_factory.create_pi_edge_api_client() zones = api_client.edge_cloud_zones() for zone in zones: zone['_id'] = zone.get('edgeCloudZoneId') insert_zones(zones) class EdgeCloudZone(BaseModel): edgeCloudZoneId: str = Field(..., description="Unique identifier of the Edge Cloud Zone") Loading
edge_cloud_management_api/controllers/federation_manager_controller.py +39 −48 Original line number Diff line number Diff line from flask import request, jsonify, Response import json from flask import request, jsonify import logging import connexion from requests.exceptions import Timeout, ConnectionError from edge_cloud_management_api.managers.log_manager import logger import requests from edge_cloud_management_api.services.federation_services import FederationManagerClientFactory # Factory pattern Loading @@ -8,74 +12,61 @@ factory = FederationManagerClientFactory() federation_client = factory.create_federation_client() def create_federation(): """POST /partner - Create federation with partner OP.""" try: body = request.get_json() result = federation_client.post_partner(body) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 body = request.get_json() token = __get_token() response = federation_client.post_partner(body, token) return jsonify(response) def get_federation(federationContextId): """GET /{federationContextId}/partner - Get federation info.""" try: result = federation_client.get_partner(federationContextId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.get_partner(federationContextId, token) return jsonify(result) def delete_federation(federationContextId): """DELETE /{federationContextId}/partner - Delete federation.""" try: result = federation_client.delete_partner(federationContextId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.delete_partner(federationContextId, token) return jsonify(result) def get_federation_context_ids(): """GET /fed-context-id - Fetch federationContextId(s).""" try: result = federation_client.get_federation_context_ids() return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() response = federation_client.get_federation_context_ids(token) return jsonify(response) def onboard_application_to_partner(federationContextId): """POST /{federationContextId}/application/onboarding - Onboard app.""" try: body = request.get_json() result = federation_client.onboard_application(federationContextId, body) return jsonify(result), 202 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.onboard_application(federationContextId, body, token) return jsonify(result) def get_onboarded_app(federationContextId, appId): """GET /{federationContextId}/application/onboarding/app/{appId}""" try: result = federation_client.get_onboarded_app(federationContextId, appId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.get_onboarded_app(federationContextId, appId, token) return jsonify(result) def delete_onboarded_app(federationContextId, appId): """DELETE /{federationContextId}/application/onboarding/app/{appId}""" try: result = federation_client.delete_onboarded_app(federationContextId, appId) return jsonify(result), 200 except Exception as e: logging.error(e.args) return jsonify(result), 500 token = __get_token() result = federation_client.delete_onboarded_app(federationContextId, appId, token) return jsonify(result) def __get_token(): bearer = connexion.request.headers['Authorization'] token = bearer.split()[1] return token
edge_cloud_management_api/controllers/network_functions_controller.py +1 −1 Original line number Diff line number Diff line from flask import jsonify from pydantic import ValidationError #Field from edge_cloud_management_api.managers.log_manager import logger from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory from edge_cloud_management_api.services.edge_cloud_services import PiEdgeAPIClientFactory def create_qod_session(body: dict): """ Loading
edge_cloud_management_api/controllers/security_controller.py 0 → 100644 +18 −0 Original line number Diff line number Diff line import os from jose import JWTError, jwt from werkzeug.exceptions import Unauthorized def decode_token(token:str): JWT_ISSUER = os.environ.get('JWT_ISSUER') PUBLIC_KEY = os.environ.get('JWT_PUBLIC_KEY') try: return jwt.decode(token, key=PUBLIC_KEY, algorithms=["RS256"], issuer=JWT_ISSUER) except JWTError as e: raise Unauthorized from e def check_oAuth2ClientCredentials(token): return {'scopes': ['fed-mgmt'], 'uid': 'test_value'} def validate_scope_oAuth2ClientCredentials(required_scopes, token_scopes): return set(required_scopes).issubset(set(token_scopes))
