# 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, operator from typing import Dict, List 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 common.orm.HighLevel import get_related_objects from .ContextModel import ContextModel LOGGER = logging.getLogger(__name__) class TopologyModel(Model): pk = PrimaryKeyField() context_fk = ForeignKeyField(ContextModel) topology_uuid = StringField(required=True, allow_empty=False) def dump_id(self) -> Dict: context_id = ContextModel(self.database, self.context_fk).dump_id() return { 'context_id': context_id, 'topology_uuid': {'uuid': self.topology_uuid}, } def dump_device_ids(self) -> List[Dict]: from .RelationModels import TopologyDeviceModel # pylint: disable=import-outside-toplevel db_devices = get_related_objects(self, TopologyDeviceModel, 'device_fk') return [db_device.dump_id() for db_device in sorted(db_devices, key=operator.attrgetter('pk'))] def dump_link_ids(self) -> List[Dict]: from .RelationModels import TopologyLinkModel # pylint: disable=import-outside-toplevel db_links = get_related_objects(self, TopologyLinkModel, 'link_fk') return [db_link.dump_id() for db_link in sorted(db_links, key=operator.attrgetter('pk'))] def dump( # pylint: disable=arguments-differ self, include_devices=True, include_links=True ) -> Dict: result = {'topology_id': self.dump_id()} if include_devices: result['device_ids'] = self.dump_device_ids() if include_links: result['link_ids'] = self.dump_link_ids() return result