Commit 48748aa9 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Refactor TelemetryFrontend to reflect Generic DB changes.

parent 56e5fd59

src/kpi_manager/database/new_KpiDB.py

+13 −13
Original line number Diff line number Diff line
@@ -14,8 +14,8 @@ 


import logging



from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

from kpi_manager.database.KpiModel import Kpi as KpiModel

from common.method_wrappers.Decorator import MetricsPool

from kpi_manager.database.KpiModel import Kpi as Model

from common.tools.database.GenericDatabase import Database

from common.method_wrappers.ServiceExceptions import OperationFailedException
@@ -25,7 +25,7 @@ METRICS_POOL = MetricsPool('KpiManager', 'Database') 
class KpiDB(Database):

    def __init__(self) -> None:

        LOGGER.info('Init KpiManagerService')

        super().__init__("tfs_kpi_mgmt", KpiModel)

        super().__init__("tfs_kpi_mgmt", Model)



    def add_row_to_db(self, row):

        return super().add_row_to_db(row)
@@ -45,33 +45,33 @@ class KpiDB(Database): 
        """

        session = self.Session()

        try:

            query = session.query(KpiModel)

            query = session.query(Model)

            # Apply filters based on the filter_object

            if filter_object.kpi_id:

                query = query.filter(KpiModel.kpi_id.in_([k.kpi_id.uuid for k in filter_object.kpi_id]))

                query = query.filter(Model.kpi_id.in_([k.kpi_id.uuid for k in filter_object.kpi_id]))



            if filter_object.kpi_sample_type:

                query = query.filter(KpiModel.kpi_sample_type.in_(filter_object.kpi_sample_type))

                query = query.filter(Model.kpi_sample_type.in_(filter_object.kpi_sample_type))



            if filter_object.device_id:

                query = query.filter(KpiModel.device_id.in_([d.device_uuid.uuid for d in filter_object.device_id]))

                query = query.filter(Model.device_id.in_([d.device_uuid.uuid for d in filter_object.device_id]))



            if filter_object.endpoint_id:

                query = query.filter(KpiModel.endpoint_id.in_([e.endpoint_uuid.uuid for e in filter_object.endpoint_id]))

                query = query.filter(Model.endpoint_id.in_([e.endpoint_uuid.uuid for e in filter_object.endpoint_id]))



            if filter_object.service_id:

                query = query.filter(KpiModel.service_id.in_([s.service_uuid.uuid for s in filter_object.service_id]))

                query = query.filter(Model.service_id.in_([s.service_uuid.uuid for s in filter_object.service_id]))



            if filter_object.slice_id:

                query = query.filter(KpiModel.slice_id.in_([s.slice_uuid.uuid for s in filter_object.slice_id]))

                query = query.filter(Model.slice_id.in_([s.slice_uuid.uuid for s in filter_object.slice_id]))



            if filter_object.connection_id:

                query = query.filter(KpiModel.connection_id.in_([c.connection_uuid.uuid for c in filter_object.connection_id]))

                query = query.filter(Model.connection_id.in_([c.connection_uuid.uuid for c in filter_object.connection_id]))



            if filter_object.link_id:

                query = query.filter(KpiModel.link_id.in_([l.link_uuid.uuid for l in filter_object.link_id]))

                query = query.filter(Model.link_id.in_([l.link_uuid.uuid for l in filter_object.link_id]))

        except Exception as e:

            LOGGER.error(f"Error creating filter of {model.__name__} table. ERROR: {e}")

            raise OperationFailedException ("CreateKpiDescriptorFilter", extra_details=["unable to create the filter {:}".format(e)]) 

        

        return super().select_with_filter(query, session, KpiModel)
 
        return super().select_with_filter(query, session, Model)

src/telemetry/database/new_Telemetry_DB.py

0 → 100644
+57 −0
Original line number Diff line number Diff line 
# Copyright 2022-2024 ETSI OSG/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 logging



from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

from .TelemetryModel import Collector as Model

from common.tools.database.GenericDatabase import Database

from common.method_wrappers.ServiceExceptions import OperationFailedException



LOGGER = logging.getLogger(__name__)

METRICS_POOL = MetricsPool('TelemteryFrontend', 'Database')



class TelemetryDB(Database):

    def __init__(self) -> None:

        LOGGER.info('Init KpiManagerService')

        super().__init__("tfs_telemetry", Model)



    def add_row_to_db(self, row):

        return super().add_row_to_db(row)



    def search_db_row_by_id(self, model, col_name, id_to_search):

        return super().search_db_row_by_id(model, col_name, id_to_search)



    def delete_db_row_by_id(self, model, col_name, id_to_search):

        return super().delete_db_row_by_id(model, col_name, id_to_search)

    

    def select_with_filter(self, model, filter_object):

        """

        Generic method to create filters dynamically based on filter_object attributes.

        params:     model:         SQLAlchemy model class to query.

                    filter_object: Object that contains filtering criteria as attributes.

        return:     SQLAlchemy session, query and Model

        """

        session = self.Session()

        try:

            query = session.query(Model)

            # Apply filters based on the filter_object

            if filter_object.kpi_id:

                query = query.filter(Model.kpi_id.in_([k.kpi_id.uuid for k in filter_object.kpi_id]))     

            # query should be added to return all rows

        except Exception as e:

            LOGGER.error(f"Error creating filter of {model.__name__} table. ERROR: {e}")

            raise OperationFailedException ("CreateKpiDescriptorFilter", extra_details=["unable to create the filter {:}".format(e)]) 

        

        return super().select_with_filter(query, session, Model)

src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py

+2 −1
Original line number Diff line number Diff line
@@ -25,7 +25,8 @@ from common.proto.telemetry_frontend_pb2 import CollectorId, Collector, Collecto 
from common.proto.telemetry_frontend_pb2_grpc import TelemetryFrontendServiceServicer



from telemetry.database.TelemetryModel import Collector as CollectorModel

from telemetry.database.Telemetry_DB import TelemetryDB

# from telemetry.database.Telemetry_DB import TelemetryDB

from ...database.new_Telemetry_DB import TelemetryDB



from confluent_kafka import Consumer as KafkaConsumer

from confluent_kafka import Producer as KafkaProducer

src/telemetry/tests/test_telemetryDB.py

+3 −3
Original line number Diff line number Diff line
@@ -14,15 +14,15 @@ 




import logging

from telemetry.database.Telemetry_DB import TelemetryDB

from telemetry.database.new_Telemetry_DB import TelemetryDB



LOGGER = logging.getLogger(__name__)



def test_verify_databases_and_tables():

    LOGGER.info('>>> test_verify_databases_and_tables : START <<< ')

    TelemetryDBobj = TelemetryDB()

    TelemetryDBobj.drop_database()

    TelemetryDBobj.verify_tables()

    # TelemetryDBobj.drop_database()

    # TelemetryDBobj.verify_tables()

    TelemetryDBobj.create_database()

    TelemetryDBobj.create_tables()

    TelemetryDBobj.verify_tables()
 
 No newline at end of file