Commit 9baba9a3 authored by George Papathanail's avatar George Papathanail
Browse files

update app_controller

parent ed1acfe5
Loading
Loading
Loading
Loading
Loading
+20 −23
Original line number Diff line number Diff line
import uuid
from flask import jsonify, request
from pydantic import ValidationError
from edge_cloud_management_api.managers.db_manager import MongoManager
from edge_cloud_management_api.managers.log_manager import logger
from edge_cloud_management_api.models.application_models import AppManifest, AppZones, AppInstance
from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory
from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClient



class NotFound404Exception(Exception):
@@ -32,7 +29,7 @@ def submit_app(body: dict):
        )


def get_apps(x_correlator=None):  # noqa: E501
def get_apps(x_correlator=None):
    """Retrieve metadata information of all applications"""
    try:
        pi_edge_factory = PiEdgeAPIClientFactory()
@@ -46,7 +43,7 @@ def get_apps(x_correlator=None): # noqa: E501
        )


def get_app(appId, x_correlator=None):  # noqa: E501
def get_app(appId, x_correlator=None):
    """Retrieve the information of an Application"""
    try:
        pi_edge_factory = PiEdgeAPIClientFactory()
@@ -67,7 +64,7 @@ def get_app(appId, x_correlator=None): # noqa: E501
        )


def delete_app(appId, x_correlator=None):  # noqa: E501
def delete_app(appId, x_correlator=None):
    """Delete Application metadata from an Edge Cloud Provider"""
    try:
        pi_edge_factory = PiEdgeAPIClientFactory()
@@ -105,7 +102,7 @@ def create_app_instance():
       
       logger.info(f"Preparing to send deployment request to SRM for appId={app_id}")
       
       pi_edge_client_factory = PiEdgeAPICLientFactory()
       pi_edge_client_factory = PiEdgeAPIClientFactory()
       pi_edge_client = pi_edge_client_factory.create_pi_edge_api_client()
       
       print("\n === Preparing Deployment Request ===")
@@ -115,7 +112,7 @@ def create_app_instance():
       print("=== End of Deployment Request ===\n")
       
       try:
          response = pi_edge_client_deploy_service_function(data=body)
          response = pi_edge_client.deploy_service_function(data=body)
          
          if isinstance(response, dict) and "error" in response:
              logger.warning(f"Failed to deploy service function: {response}")
@@ -178,7 +175,7 @@ def delete_app_instance(appInstanceId: str, x_correlator=None):
        pi_edge_client_factory = PiEdgeAPIClientFactory()
        pi_edge_client = pi_edge_client_factory.create_pi_edge_api_client()
        response = pi_edge_client.delete_app_instance(appInstanceId)
        return jsonify({'result': response.text, 'status': response.status_code})
        return response.json()

    except Exception as e:
        return (
+4 −5
Original line number Diff line number Diff line
@@ -41,10 +41,11 @@ def get_local_zones() -> list[dict]:
            logger.error(f"SRM error: {result['error']}")
            return []


        return result

    except Exception as e:
        logger.exception("Unexpected error while retrieving local zones from SRM")
        logger.exception("Unexpected error while retrieving local zones from SRM: %s", e)
        return []


@@ -58,10 +59,10 @@ def get_all_cloud_zones() -> List[EdgeCloudZone]:
    return get_local_zones() + get_federated_zones()


def get_edge_cloud_zones(x_correlator: str | None = None, region=None, status=None):
def get_edge_cloud_zones(x_correlator: str | None = None, region=None, status=None):  # noqa: E501
    """Retrieve a list of the operators Edge Cloud Zones and their status

    List of the operators Edge Cloud Zones and their status, ordering the results by location and filtering by status (active/inactive/unknown)
    List of the operators Edge Cloud Zones and their status, ordering the results by location and filtering by status (active/inactive/unknown)  # noqa: E501

    :param x_correlator: Correlation id for the different services
    :type x_correlator: str
@@ -87,8 +88,6 @@ def get_edge_cloud_zones(x_correlator: str | None = None, region=None, status=No
            """If status is None, return True (don't apply status filtering), otherwise check if the zone status matches the query status"""
            return (query_params.status is None) or (zone["edgeCloudZoneStatus"] == query_params.status)

        # filtered_zones = [zone for zone in get_all_cloud_zones() if (query_region_matches(zone) and query_status_matches(zone))]

        response = [EdgeCloudZone(**zone).model_dump() for zone in get_all_cloud_zones()]
        return jsonify(response), 200