add baseline implementation of CE server

.dockerignore

+.travis.yaml
+.swagger-codegen-ignore
+README.md
+tox.ini
+git_push.sh
+test-requirements.txt
+setup.py
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+venv/
+.python-version
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+#Ipython Notebook
+.ipynb_checkpoints

.env.sample

+PI_EDGE_BASE_URL=http://example.com/api
+PI_EDGE_USERNAME=username
+PI_EDGE_PASSWORD=password
+HTTP_PROXY=https://company.proxy:3128

.gitignore

+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+venv/
+.python-version
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+#Ipython Notebook
+.ipynb_checkpoints
+
+# virtual environment folders
+.venv
+
+# environment
+.env
+
+# code editors
+.vscode/
\ No newline at end of file

.swagger-codegen-ignore

+# Swagger Codegen Ignore
+# Generated by swagger-codegen https://github.com/swagger-api/swagger-codegen
+
+# Use this file to prevent files from being overwritten by the generator.
+# The patterns follow closely to .gitignore or .dockerignore.
+
+# As an example, the C# client generator defines ApiClient.cs.
+# You can make changes and tell Swagger Codgen to ignore just this file by uncommenting the following line:
+#ApiClient.cs
+
+# You can match any string of characters against a directory, file or extension with a single asterisk (*):
+#foo/*/qux
+# The above matches foo/bar/qux and foo/baz/qux, but not foo/bar/baz/qux
+
+# You can recursively match patterns against a directory, file or extension with a double asterisk (**):
+#foo/**/qux
+# This matches foo/bar/qux, foo/baz/qux, and foo/bar/baz/qux
+
+# You can also negate patterns with an exclamation (!).
+# For example, you can ignore all files in a docs folder with the file extension .md:
+#docs/*.md
+# Then explicitly reverse the ignore rule for a single file:
+#!docs/README.md

.swagger-codegen/VERSION

+3.0.63
\ No newline at end of file

.travis.yml

+# ref: https://docs.travis-ci.com/user/languages/python
+language: python
+python:
+  - "3.10"
+  - "3.11"
+  - "3.12"
+install: "pip install -r requirements.txt"
+# command to run tests
+script: nosetests

Dockerfile

+FROM python:3.12-alpine
+
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+COPY requirements.txt /usr/src/app/
+
+RUN pip3 install --no-cache-dir --trusted-host pypi.org --trusted-host files.pythonhosted.org -r requirements.txt
+
+COPY . /usr/src/app
+
+EXPOSE 8080
+
+ENTRYPOINT ["python3"]
+
+CMD ["-m", "swagger_server"]
\ No newline at end of file

README.md

-# capabilities-exposure-api
-CAMARA compliant Capabilities Exposure API
+# Capabilities Exposure API Server
+
+A Server Implementation of the **CAMARA Edge Application Management API** as specified [here](https://github.com/camaraproject/EdgeCloud/blob/main/code/API_definitions/Edge-Application-Management.yaml)
+
+
+### Overview
+This server was generated by the [swagger-codegen](https://github.com/swagger-api/swagger-codegen) project.
+It uses the [Connexion](https://github.com/zalando/connexion) library on top of Flask.
+It propagates requests to the [Sunrise 6G OP service resource manager](https://github.com/SMARTECH-ISI-Athena/Sunrise-6G-Operator-Platform), as configured in the `.env`.
+
+### Usage
+Before running the server you need to `mv env.sample .env` and update variables to properly link to a running OP service resource manager instance.
+
+To run the server, please execute the following from the root directory:
+
+```
+pip3 install -r requirements.txt
+python3 -m swagger_server
+```
+
+and open your browser to the OpenAPI documentation:
+
+```
+http://localhost:8080/edge-application-management/ui/
+```
+
+Your OpenAPI definition lives here:
+
+```
+http://localhost:8080/edge-application-management/openapi.json
+```
+
+To launch the integration tests, use tox:
+```
+sudo pip install tox
+tox
+```
+
+### Running with Docker
+
+To run the server on a Docker container, please execute the following from the root directory:
+
+```bash
+# building the image
+docker build -t swagger_server .
+
+# starting up a container
+docker run -p 8080:8080 swagger_server
+```
\ No newline at end of file

requirements.txt

+connexion >= 2.6.0
+connexion[swagger-ui] >= 2.6.0
+python_dateutil == 2.6.0
+setuptools >= 21.0.0
+swagger-ui-bundle >= 0.0.2
+python-dotenv == 1.0.1
\ No newline at end of file

setup.py

+# coding: utf-8
+
+import sys
+from setuptools import setup, find_packages
+
+NAME = "swagger_server"
+VERSION = "1.0.0"
+# To install the library, run the following
+#
+# python setup.py install
+#
+# prerequisite: setuptools
+# http://pypi.python.org/pypi/setuptools
+
+REQUIRES = [
+    "connexion",
+    "swagger-ui-bundle>=0.0.2"
+]
+
+setup(
+    name=NAME,
+    version=VERSION,
+    description="Capabilities Exposure API Server",
+    author_email="dkaragkounis@intracom-telecom.com",
+    url="",
+    keywords=["Swagger", "Capabilities Exposure API Server"],
+    install_requires=REQUIRES,
+    packages=find_packages(),
+    package_data={'': ['swagger/swagger.yaml']},
+    include_package_data=True,
+    entry_points={
+        'console_scripts': ['swagger_server=swagger_server.__main__:main']},
+    long_description="""\
+    The Capabilities Exposure API Server allows API consumers (Application Providers) to manage the Life Cycle of an Application and to Discover Edge Cloud Zones. 
+    """
+)

swagger_server/__main__.py

+#!/usr/bin/env python3
+
+import connexion
+
+from swagger_server import encoder
+
+
+def main():
+    # app = connexion.App(__name__, specification_dir='./swagger/')
+    app = connexion.FlaskApp(__name__)
+    app.app.json_encoder = encoder.JSONEncoder
+    app.add_api('./swagger/swagger.yaml', arguments={'title': 'Capabilities Exposure API Server'}, pythonic_params=True)
+    app.run(port=8080)
+
+
+if __name__ == '__main__':
+    main()

swagger_server/configs/env_config.py

+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+PI_EDGE_BASE_URL = os.getenv("PI_EDGE_BASE_URL")
+PI_EDGE_USERNAME = os.getenv("PI_EDGE_USERNAME")
+PI_EDGE_PASSWORD = os.getenv("PI_EDGE_PASSWORD")
+HTTP_PROXY = os.getenv("HTTP_PROXY")

swagger_server/configs/logger_config.py

+import logging
+import sys
+
+# logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("Capabilities Exposure API Server")
+logger.setLevel(logging.INFO)
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+handler = logging.StreamHandler(sys.stdout)
+handler.setFormatter(formatter)

swagger_server/controllers/application_controller.py

+import connexion
+import six
+
+from swagger_server.configs.logger_config import logger
+from swagger_server.filters.service_functions import find_requested_service_function
+from swagger_server.models.app_id import AppId  # noqa: E501
+from swagger_server.models.app_instance_id import AppInstanceId  # noqa: E501
+from swagger_server.models.app_manifest import AppManifest  # noqa: E501
+from swagger_server.models.edge_cloud_region import EdgeCloudRegion  # noqa: E501
+from swagger_server.models.edge_cloud_zone import EdgeCloudZone  # noqa: E501
+from swagger_server.models.error_info import ErrorInfo  # noqa: E501
+from swagger_server.models.inline_response200 import InlineResponse200  # noqa: E501
+from swagger_server.models.inline_response2001 import InlineResponse2001  # noqa: E501
+from swagger_server.models.inline_response202 import InlineResponse202  # noqa: E501
+from swagger_server.models.submitted_app import SubmittedApp  # noqa: E501
+from swagger_server import util
+from swagger_server.schema_mappers.application_instantiation_mapper import (
+    map_to_pi_edge_deployed_service_function,
+)
+from swagger_server.services.pi_edge_services import PiEdgeAPIClientFactory
+
+
+def create_app_instance(body, app_id, x_correlator=None):  # noqa: E501
+    """Instantiation of an Application
+
+    Ask the Edge Cloud Platform to instantiate an application to one or several Edge Cloud Zones with an Application as an input and an Application Instance as the output.  # noqa: E501
+
+    :param body: Array of Edge Cloud Zone
+    :type body: list | bytes
+    :param app_id: A globally unique identifier associated with the application. Edge Cloud Provider generates this identifier when the application is submitted.
+    :type app_id: dict | bytes
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+
+    :rtype: InlineResponse202
+    """
+    try:
+        if connexion.request.is_json:
+            body = [
+                EdgeCloudZone.from_dict(d) for d in connexion.request.get_json()
+            ]  # noqa: E501
+        # if connexion.request.is_json:
+        #     app_id = AppId.from_dict(connexion.request.get_json())  # noqa: E501
+
+        logger.critical(body)
+        logger.critical(f"Test: {app_id}")
+
+        pi_edge_factory = PiEdgeAPIClientFactory()
+        pi_edge_api_client = pi_edge_factory.create_pi_edge_api_client()
+
+        service_functions_catalogue = (
+            pi_edge_api_client.get_service_functions_catalogue()
+        )
+        requested_application = find_requested_service_function(
+            service_functions_catalogue, app_id
+        )
+        if requested_application is None:
+            return {
+                "status": 400,
+                "code": "INVALID_ARGUMENT",
+                "message": "The requested appId could not be retrieved. Please verify application metadata for the provided id do exists.",
+            }, 400
+
+        deployed_service_functions = [
+            map_to_pi_edge_deployed_service_function(
+                edge_cloud, requested_application["name"]
+            )
+            for edge_cloud in body
+        ]
+
+        for service_function in deployed_service_functions:
+            result = pi_edge_api_client.deploy_service_function(data=service_function)
+            if result != None:  # an error occurred
+                if "error" in result.keys():
+                    raise Exception(result["error"])
+                else:
+                    raise Exception("Unexpected error in app instantiation")
+        return {}, 202  # application instantiation accepted
+    except:
+        logger.exception("Error instantiating application")
+        return {
+            "status": 500,
+            "code": "INTERNAL",
+            "message": "Internal server error.",
+        }, 500
+
+
+def delete_app(app_id, x_correlator=None):  # noqa: E501
+    """Delete an Application from an Edge Cloud Provider
+
+    Delete all the information and content related to an Application # noqa: E501
+
+    :param app_id: Identificator of the application to be deleted provided by the Edge Cloud Provider once the submission was successful
+    :type app_id: dict | bytes
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+
+    :rtype: None
+    """
+    if connexion.request.is_json:
+        app_id = AppId.from_dict(connexion.request.get_json())  # noqa: E501
+    return "do some magic!"
+
+
+def delete_app_instance(app_id, app_instance_id, x_correlator=None):  # noqa: E501
+    """Terminate an Application Instance
+
+    Terminate a running instance of an application within an Edge Cloud Zone  # noqa: E501
+
+    :param app_id: A globally unique identifier associated with the application. Edge Cloud Provider generates this identifier when the application is submitted.
+    :type app_id: dict | bytes
+    :param app_instance_id: Identificator of the specific application instance that will be terminated
+    :type app_instance_id: dict | bytes
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+
+    :rtype: None
+    """
+    if connexion.request.is_json:
+        app_id = AppId.from_dict(connexion.request.get_json())  # noqa: E501
+    if connexion.request.is_json:
+        app_instance_id = AppInstanceId.from_dict(
+            connexion.request.get_json()
+        )  # noqa: E501
+    return "do some magic!"
+
+
+def get_app(app_id, x_correlator=None):  # noqa: E501
+    """Retrieve the information of an Application
+
+    Ask the Edge Cloud Provider the information for a given application  # noqa: E501
+
+    :param app_id: A globally unique identifier associated with the application. Edge Cloud Provider generates this identifier when the application is submitted.
+    :type app_id: dict | bytes
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+
+    :rtype: InlineResponse200
+    """
+    if connexion.request.is_json:
+        app_id = AppId.from_dict(connexion.request.get_json())  # noqa: E501
+    return "do some magic!"
+
+
+def get_app_instance(
+    app_id, x_correlator=None, app_instance_id=None, region=None
+):  # noqa: E501
+    """Retrieve the information of Application Instances for a given App
+
+    Ask the Edge Cloud Provider the information of the instances for a given application  # noqa: E501
+
+    :param app_id: A globally unique identifier associated with the application. Edge Cloud Provider generates this identifier when the application is submitted.
+    :type app_id: dict | bytes
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+    :param app_instance_id: A globally unique identifier associated with a running instance of an application within an specific Edge Cloud Zone. Edge Cloud Provider generates this identifier.
+    :type app_instance_id: dict | bytes
+    :param region: Human readable name of the geographical Edge Cloud Region of the Edge Cloud. Defined by the Edge Cloud Provider.
+    :type region: dict | bytes
+
+    :rtype: InlineResponse2001
+    """
+    if connexion.request.is_json:
+        app_id = AppId.from_dict(connexion.request.get_json())  # noqa: E501
+    if connexion.request.is_json:
+        app_instance_id = AppInstanceId.from_dict(
+            connexion.request.get_json()
+        )  # noqa: E501
+    if connexion.request.is_json:
+        region = EdgeCloudRegion.from_dict(connexion.request.get_json())  # noqa: E501
+    return "do some magic!"
+
+
+def submit_app(body, x_correlator=None):  # noqa: E501
+    """Submit application metadata to the Edge Cloud Provider.
+
+        Contains the information about the application to be instantiated in the Edge Cloud  # noqa: E501
+
+        :param body: The Application Provider request contains mandatory
+    criteria (e.g. required CPU, memory, storage, bandwidth) and
+    optional parameters.
+
+        :type body: dict | bytes
+        :param x_correlator: Correlation id for the different services
+        :type x_correlator: str
+
+        :rtype: SubmittedApp
+    """
+    if connexion.request.is_json:
+        body = AppManifest.from_dict(connexion.request.get_json())  # noqa: E501
+    return "do some magic!"

swagger_server/controllers/authorization_controller.py

+from typing import List
+"""
+controller generated to handled auth operation described at:
+https://connexion.readthedocs.io/en/latest/security.html
+"""
+

swagger_server/controllers/edge_cloud_controller.py

+import connexion
+import six
+
+from swagger_server.models.edge_cloud_region import EdgeCloudRegion  # noqa: E501
+from swagger_server.models.edge_cloud_zone_status import (
+    EdgeCloudZoneStatus,
+)  # noqa: E501
+from swagger_server.models.edge_cloud_zones import EdgeCloudZones  # noqa: E501
+from swagger_server.models.error_info import ErrorInfo  # noqa: E501
+from swagger_server import util
+from swagger_server.schema_mappers.edge_cloud_mapper import map_to_edge_cloud
+from swagger_server.services.pi_edge_services import PiEdgeAPIClientFactory
+
+
+def get_edge_cloud_zones(x_correlator=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)  # noqa: E501
+
+    :param x_correlator: Correlation id for the different services
+    :type x_correlator: str
+    :param region: Human readable name of the geographical Edge Cloud Region of the Edge Cloud. Defined by the Edge Cloud Provider.
+    :type region: dict | bytes
+    :param status: Human readable status of the Edge Cloud Zone
+    :type status: dict | bytes
+
+    :rtype: EdgeCloudZones
+    """
+    try:
+        if connexion.request.is_json:
+            region = EdgeCloudRegion.from_dict(
+                connexion.request.get_json()
+            )  # noqa: E501
+        if connexion.request.is_json:
+            status = EdgeCloudZoneStatus.from_dict(
+                connexion.request.get_json()
+            )  # noqa: E501
+
+        pi_edge_factory = PiEdgeAPIClientFactory()
+        pi_edge_api_client = pi_edge_factory.create_pi_edge_api_client()
+        nodes: list | None = pi_edge_api_client.edge_cloud_zones()
+
+        if not nodes:
+            error_info = ErrorInfo(
+                status=404, code="NOT_FOUND", message="No edge cloud zones found."
+            )
+            return error_info, 404
+
+        # Map nodes to EdgeCloudZones
+        edge_cloud_zones = [map_to_edge_cloud(zone) for zone in nodes]
+
+        return edge_cloud_zones, 200
+
+    except Exception as e:
+        error_info = ErrorInfo(
+            status=500, code="INTERNAL_ERROR", message=f"An error occurred: {str(e)}"
+        )
+        return error_info, 500

swagger_server/encoder.py

+from connexion.apps.flask_app import FlaskJSONEncoder
+import six
+
+from swagger_server.models.base_model_ import Model
+
+
+class JSONEncoder(FlaskJSONEncoder):
+    include_nulls = False
+
+    def default(self, o):
+        if isinstance(o, Model):
+            dikt = {}
+            for attr, _ in six.iteritems(o.swagger_types):
+                value = getattr(o, attr)
+                if value is None and not self.include_nulls:
+                    continue
+                attr = o.attribute_map[attr]
+                dikt[attr] = value
+            return dikt
+        return FlaskJSONEncoder.default(self, o)