Skip to content
Snippets Groups Projects

Polish "Implement CAMARA-based NBI connector"

6 files
+ 37
62
Compare changes
  • Side-by-side
  • Inline
Files
6
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
#
# Licensed under the Apache License, Version 2.0 (the "License");
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# you may not use this file except in compliance with the License.
@@ -11,31 +11,26 @@
@@ -11,31 +11,26 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# See the License for the specific language governing permissions and
# limitations under the License.
# limitations under the License.
from flask.json import jsonify
from flask_restful import Resource, request
from enum import Enum
import copy, grpc, grpc._channel, logging
import grpc._channel
from qos_profile.client.QoSProfileClient import QoSProfileClient
from werkzeug.exceptions import UnsupportedMediaType
from common.proto.context_pb2 import QoSProfileId, Uuid,Empty
from common.proto.qos_profile_pb2 import QoDConstraintsRequest
from typing import Dict
from typing import Dict
from uuid import uuid4
from uuid import uuid4
import grpc, logging
from flask.json import jsonify
import copy, deepmerge, json, logging
from typing import Dict
from flask_restful import Resource, request
from flask_restful import Resource, request
from werkzeug.exceptions import UnsupportedMediaType
from common.proto.context_pb2 import QoSProfileId, Uuid,Empty
from common.Constants import DEFAULT_CONTEXT_NAME
from common.Constants import DEFAULT_CONTEXT_NAME
from context.client.ContextClient import ContextClient
from context.client.ContextClient import ContextClient
 
from qos_profile.client.QoSProfileClient import QoSProfileClient
from service.client.ServiceClient import ServiceClient
from service.client.ServiceClient import ServiceClient
from .Tools import (
from .Tools import (
format_grpc_to_json, grpc_context_id, grpc_service_id, QOD_2_service, service_2_qod,grpc_message_to_qos_table_data,create_qos_profile_from_json
format_grpc_to_json, grpc_context_id, grpc_service_id,
 
QOD_2_service, service_2_qod, grpc_message_to_qos_table_data,
 
create_qos_profile_from_json
)
)
 
LOGGER = logging.getLogger(__name__)
LOGGER = logging.getLogger(__name__)
#Initiate the QoSProfileClient
class _Resource(Resource):
class _Resource(Resource):
def __init__(self) -> None:
def __init__(self) -> None:
super().__init__()
super().__init__()
@@ -43,7 +38,7 @@ class _Resource(Resource):
@@ -43,7 +38,7 @@ class _Resource(Resource):
self.client = ContextClient()
self.client = ContextClient()
self.service_client = ServiceClient()
self.service_client = ServiceClient()
#ProfileList Endpoint for posting
#ProfileList Endpoint for posting
class ProfileList(_Resource):
class ProfileList(_Resource):
def post(self):
def post(self):
if not request.is_json:
if not request.is_json:
@@ -56,13 +51,13 @@ class ProfileList(_Resource):
@@ -56,13 +51,13 @@ class ProfileList(_Resource):
except Exception as e:
except Exception as e:
LOGGER.error(e) # track if there is an error
LOGGER.error(e) # track if there is an error
raise e
raise e
# Send to gRPC server using CreateQosProfile done by Shayan
# Send to gRPC server using CreateQosProfile
try:
try:
qos_profile_created = self.qos_profile_client.CreateQoSProfile(qos_profile)
qos_profile_created = self.qos_profile_client.CreateQoSProfile(qos_profile)
except Exception as e:
except Exception as e:
LOGGER.info(e)
LOGGER.error(e)
raise e
raise e
#gRPC message back to JSON using the helper function created by shayan
#gRPC message back to JSON using the helper function
qos_profile_data = grpc_message_to_qos_table_data(qos_profile_created)
qos_profile_data = grpc_message_to_qos_table_data(qos_profile_created)
LOGGER.info(f'qos_profile_data{qos_profile_data}')
LOGGER.info(f'qos_profile_data{qos_profile_data}')
return jsonify(qos_profile_data)
return jsonify(qos_profile_data)
@@ -140,8 +135,7 @@ class ProfileDetail(_Resource):
@@ -140,8 +135,7 @@ class ProfileDetail(_Resource):
return {"error": "Internal Server Error"}, 500
return {"error": "Internal Server Error"}, 500
###SESSION##########################################################
###SESSION##########################################################
LOGGER = logging.getLogger(__name__)
class QodInfo(_Resource):
class qodinfo(_Resource):
def post(self):
def post(self):
if not request.is_json:
if not request.is_json:
return (jsonify({'error': 'Unsupported Media Type', 'message': 'JSON payload is required'}), 415)
return (jsonify({'error': 'Unsupported Media Type', 'message': 'JSON payload is required'}), 415)
@@ -163,18 +157,18 @@ class qodinfo(_Resource):
@@ -163,18 +157,18 @@ class qodinfo(_Resource):
response = format_grpc_to_json(self.service_client.CreateService(stripped_service))
response = format_grpc_to_json(self.service_client.CreateService(stripped_service))
response = format_grpc_to_json(self.service_client.UpdateService(service))
response = format_grpc_to_json(self.service_client.UpdateService(service))
except Exception as e: # pylint: disable=broad-except
except Exception as e: # pylint: disable=broad-except
LOGGER.error(f"error related to response: {response}")
LOGGER.error(f"Unexpected error: {str(e)}", exc_info=True)
return e
return jsonify({'error': 'Internal Server Error', 'message': 'An unexpected error occurred'}), 500
LOGGER.error(f"error related to response: {response}")
return response
return response
def get(self):
def get(self):
service_list = self.client.ListServices(grpc_context_id(DEFAULT_CONTEXT_NAME)) #return context id as json
service_list = self.client.ListServices(grpc_context_id(DEFAULT_CONTEXT_NAME)) #return context id as json
qod_info = [service_2_qod(service) for service in service_list.services] #iterating over service list
qod_info = [service_2_qod(service) for service in service_list.services] #iterating over service list
LOGGER.info(f"error related to qod_info: {qod_info}")
LOGGER.info(f"error related to qod_info: {qod_info}")
return qod_info
return jsonify(qod_info), 200
class qodinfoId(_Resource):
class QodInfoID(_Resource):
def get(self, sessionId: str):
def get(self, sessionId: str):
try:
try:
Loading