EndPointModel.py

# 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 enum, functools
from typing import Dict
from sqlalchemy import Column, String, Enum, ForeignKeyConstraint
from sqlalchemy.dialects.postgresql import ARRAY, UUID
from sqlalchemy.orm import relationship
from common.proto.kpi_sample_types_pb2 import KpiSampleType
from ._Base import _Base
from .Tools import grpc_to_enum

class ORM_KpiSampleTypeEnum(enum.Enum):
    UNKNOWN             = KpiSampleType.KPISAMPLETYPE_UNKNOWN
    PACKETS_TRANSMITTED = KpiSampleType.KPISAMPLETYPE_PACKETS_TRANSMITTED
    PACKETS_RECEIVED    = KpiSampleType.KPISAMPLETYPE_PACKETS_RECEIVED
    BYTES_TRANSMITTED   = KpiSampleType.KPISAMPLETYPE_BYTES_TRANSMITTED
    BYTES_RECEIVED      = KpiSampleType.KPISAMPLETYPE_BYTES_RECEIVED

grpc_to_enum__kpi_sample_type = functools.partial(
    grpc_to_enum, KpiSampleType, ORM_KpiSampleTypeEnum)

class EndPointModel(_Base):
    __tablename__ = 'endpoint'
    context_uuid  = Column(UUID(as_uuid=False), primary_key=True)
    topology_uuid = Column(UUID(as_uuid=False), primary_key=True)
    device_uuid   = Column(UUID(as_uuid=False), primary_key=True)
    endpoint_uuid = Column(UUID(as_uuid=False), primary_key=True)
    endpoint_type = Column(String)
    kpi_sample_types = Column(ARRAY(Enum(ORM_KpiSampleTypeEnum), dimensions=1))

    __table_args__ = (
        ForeignKeyConstraint(
            ['context_uuid', 'topology_uuid'],
            ['topology.context_uuid', 'topology.topology_uuid'],
            ondelete='CASCADE'),
        ForeignKeyConstraint(
            ['device_uuid'],
            ['device.device_uuid'],
            ondelete='CASCADE'),
    )

    topology = relationship('TopologyModel', back_populates='endpoints')
    device   = relationship('DeviceModel', back_populates='endpoints')

    def dump_id(self) -> Dict:
        result = {
            'topology_id': self.topology.dump_id(),
            'device_id': self.device.dump_id(),
            'endpoint_uuid': {'uuid': self.endpoint_uuid},
        }
        return result

    def dump(self) -> Dict:
        return {
            'endpoint_id'     : self.dump_id(),
            'endpoint_type'   : self.endpoint_type,
            'kpi_sample_types': [kst.value for kst in self.kpi_sample_types],
        }

# def get_endpoint(
#     database : Database, grpc_endpoint_id : EndPointId,
#     validate_topology_exists : bool = True, validate_device_in_topology : bool = True
# ) -> Tuple[str, EndPointModel]:
#     endpoint_uuid                  = grpc_endpoint_id.endpoint_uuid.uuid
#     endpoint_device_uuid           = grpc_endpoint_id.device_id.device_uuid.uuid
#     endpoint_topology_uuid         = grpc_endpoint_id.topology_id.topology_uuid.uuid
#     endpoint_topology_context_uuid = grpc_endpoint_id.topology_id.context_id.context_uuid.uuid
#     str_endpoint_key = key_to_str([endpoint_device_uuid, endpoint_uuid])
#
#     if len(endpoint_topology_context_uuid) > 0 and len(endpoint_topology_uuid) > 0:
#         # check topology exists
#         str_topology_key = key_to_str([endpoint_topology_context_uuid, endpoint_topology_uuid])
#         if validate_topology_exists:
#             from .TopologyModel import TopologyModel
#             get_object(database, TopologyModel, str_topology_key)
#
#         # check device is in topology
#         str_topology_device_key = key_to_str([str_topology_key, endpoint_device_uuid], separator='--')
#         if validate_device_in_topology:
#             from .RelationModels import TopologyDeviceModel
#             get_object(database, TopologyDeviceModel, str_topology_device_key)
#
#         str_endpoint_key = key_to_str([str_endpoint_key, str_topology_key], separator=':')
#
#     db_endpoint : EndPointModel = get_object(database, EndPointModel, str_endpoint_key)
#     return str_endpoint_key, db_endpoint