# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 typing import Dict, List from common.orm.Database import Database from common.orm.backend.Tools import key_to_str from common.orm.fields.EnumeratedField import EnumeratedField from common.orm.fields.ForeignKeyField import ForeignKeyField from common.orm.fields.PrimaryKeyField import PrimaryKeyField from common.orm.fields.StringField import StringField from common.orm.model.Model import Model from .DeviceModel import DeviceModel from .KpiSampleType import ORM_KpiSampleTypeEnum, grpc_to_enum__kpi_sample_type from .TopologyModel import TopologyModel LOGGER = logging.getLogger(__name__) class EndPointModel(Model): pk = PrimaryKeyField() topology_fk = ForeignKeyField(TopologyModel, required=False) device_fk = ForeignKeyField(DeviceModel) endpoint_uuid = StringField(required=True, allow_empty=False) endpoint_type = StringField() resource_key = StringField(required=True, allow_empty=False) def dump_id(self) -> Dict: device_id = DeviceModel(self.database, self.device_fk).dump_id() result = { 'device_id': device_id, 'endpoint_uuid': {'uuid': self.endpoint_uuid}, } if self.topology_fk is not None: result['topology_id'] = TopologyModel(self.database, self.topology_fk).dump_id() return result def dump_kpi_sample_types(self) -> List[int]: db_kpi_sample_type_pks = self.references(EndPointMonitorModel) return [EndPointMonitorModel(self.database, pk).dump() for pk,_ in db_kpi_sample_type_pks] def dump( # pylint: disable=arguments-differ self, include_kpi_sample_types=True ) -> Dict: result = { 'endpoint_id': self.dump_id(), 'endpoint_type': self.endpoint_type, } if include_kpi_sample_types: result['kpi_sample_types'] = self.dump_kpi_sample_types() return result class EndPointMonitorModel(Model): # pylint: disable=abstract-method pk = PrimaryKeyField() endpoint_fk = ForeignKeyField(EndPointModel) resource_key = StringField(required=True, allow_empty=True) kpi_sample_type = EnumeratedField(ORM_KpiSampleTypeEnum, required=True) def dump(self) -> Dict: return self.kpi_sample_type.value def set_endpoint_monitors(database : Database, db_endpoint : EndPointModel, grpc_endpoint_kpi_sample_types): db_endpoint_pk = db_endpoint.pk for kpi_sample_type in grpc_endpoint_kpi_sample_types: orm_kpi_sample_type = grpc_to_enum__kpi_sample_type(kpi_sample_type) str_endpoint_kpi_sample_type_key = key_to_str([db_endpoint_pk, orm_kpi_sample_type.name]) db_endpoint_kpi_sample_type = EndPointMonitorModel(database, str_endpoint_kpi_sample_type_key) db_endpoint_kpi_sample_type.endpoint_fk = db_endpoint db_endpoint_kpi_sample_type.resource_key = '' # during initialization, allow empty value db_endpoint_kpi_sample_type.kpi_sample_type = orm_kpi_sample_type db_endpoint_kpi_sample_type.save()