OsmClientServiceServicerImpl.py

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import grpc, logging
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.tools.grpc.Tools import grpc_message_to_json_string
from common.proto.context_pb2 import (Empty)
from common.proto.osm_client_pb2 import CreateRequest, CreateResponse, NsiListResponse, GetRequest, GetResponse, DeleteRequest, DeleteResponse
from common.proto.osm_client_pb2_grpc import OsmServiceServicer
from osmclient import client
from osmclient.common.exceptions import ClientException
from osm_client.Config import OSM_ADDRESS

LOGGER = logging.getLogger(__name__)

METRICS_POOL = MetricsPool('OSMCLIENT', 'RPC')

class OsmClientServiceServicerImpl(OsmServiceServicer):
    def __init__(self):
        LOGGER.info('Creating Servicer...')
        self.myclient = client.Client(host=OSM_ADDRESS, sol005=True)
        LOGGER.info('osmClient created')

        LOGGER.info('Servicer Created')

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def NsiCreate(self, request : CreateRequest, context : grpc.ServicerContext) -> CreateResponse:
        try:
            #OSM library doesn't return nsi ID, just an exception
            self.myclient.nsi.create(request.nst_name, request.nsi_name, request.account)
        except Exception as e:
            resp = CreateResponse(succeded = False, errormessage = str(e))
        else:
            resp =  CreateResponse(succeded = True)
        return resp
    
    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def NsiList(self, request : Empty, context : grpc.ServicerContext) -> NsiListResponse:
        nsiIDs = self.myclient.nsi.list()
        resp = NsiListResponse(id=nsiIDs)
        return resp


    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def NsiGet(self, request : GetRequest, context : grpc.ServicerContext) -> GetResponse:
        nsiObject =  self.myclient.nsi.get(request.id)
        resp = GetResponse(NsiObject = nsiObject)
        return resp

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def NsiDelete(self, request : DeleteRequest, context : grpc.ServicerContext) -> DeleteResponse:
        try:
            #OSM library doesn't return nsi ID, just an exception
            self.myclient.nsi.delete(request.id, False, False)
        except Exception as e:
            resp = DeleteResponse(succeded = False, errormessage = str(e))
        else:
            resp = DeleteResponse(succeded = True)
        return resp