# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (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 datetime, logging from sqlalchemy.dialects.postgresql import insert from sqlalchemy.engine import Engine from sqlalchemy.orm import Session, selectinload, sessionmaker from sqlalchemy_cockroachdb import run_transaction from typing import Dict, List, Optional, Set, Tuple from common.proto.context_pb2 import Link, LinkId from common.method_wrappers.ServiceExceptions import NotFoundException from common.tools.object_factory.Link import json_link_id from .models.LinkModel import LinkModel, LinkEndPointModel from .models.TopologyModel import TopologyLinkModel from .uuids.EndPoint import endpoint_get_uuid from .uuids.Link import link_get_uuid LOGGER = logging.getLogger(__name__) def link_list_ids(db_engine : Engine) -> List[Dict]: def callback(session : Session) -> List[Dict]: obj_list : List[LinkModel] = session.query(LinkModel).all() return [obj.dump_id() for obj in obj_list] return run_transaction(sessionmaker(bind=db_engine), callback) def link_list_objs(db_engine : Engine) -> List[Dict]: def callback(session : Session) -> List[Dict]: obj_list : List[LinkModel] = session.query(LinkModel)\ .options(selectinload(LinkModel.link_endpoints))\ .all() return [obj.dump() for obj in obj_list] return run_transaction(sessionmaker(bind=db_engine), callback) def link_get(db_engine : Engine, request : LinkId) -> Dict: link_uuid = link_get_uuid(request, allow_random=False) def callback(session : Session) -> Optional[Dict]: obj : Optional[LinkModel] = session.query(LinkModel)\ .options(selectinload(LinkModel.link_endpoints))\ .filter_by(link_uuid=link_uuid).one_or_none() return None if obj is None else obj.dump() obj = run_transaction(sessionmaker(bind=db_engine), callback) if obj is None: raw_link_uuid = request.link_uuid.uuid raise NotFoundException('Link', raw_link_uuid, extra_details=[ 'link_uuid generated was: {:s}'.format(link_uuid) ]) return obj def link_set(db_engine : Engine, request : Link) -> Tuple[Dict, bool]: raw_link_uuid = request.link_id.link_uuid.uuid raw_link_name = request.name link_name = raw_link_uuid if len(raw_link_name) == 0 else raw_link_name link_uuid = link_get_uuid(request.link_id, link_name=link_name, allow_random=True) now = datetime.datetime.utcnow() topology_uuids : Set[str] = set() related_topologies : List[Dict] = list() link_endpoints_data : List[Dict] = list() for i,endpoint_id in enumerate(request.link_endpoint_ids): endpoint_topology_uuid, _, endpoint_uuid = endpoint_get_uuid( endpoint_id, allow_random=False) link_endpoints_data.append({ 'link_uuid' : link_uuid, 'endpoint_uuid': endpoint_uuid, 'position' : i, }) if endpoint_topology_uuid not in topology_uuids: related_topologies.append({ 'topology_uuid': endpoint_topology_uuid, 'link_uuid' : link_uuid, }) topology_uuids.add(endpoint_topology_uuid) link_data = [{ 'link_uuid' : link_uuid, 'link_name' : link_name, 'created_at': now, 'updated_at': now, }] def callback(session : Session) -> bool: stmt = insert(LinkModel).values(link_data) stmt = stmt.on_conflict_do_update( index_elements=[LinkModel.link_uuid], set_=dict( link_name = stmt.excluded.link_name, updated_at = stmt.excluded.updated_at, ) ) stmt = stmt.returning(LinkModel.created_at, LinkModel.updated_at) created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at if len(link_endpoints_data) > 0: # TODO: manage add/remove of endpoints; manage changes in relations with topology stmt = insert(LinkEndPointModel).values(link_endpoints_data) stmt = stmt.on_conflict_do_nothing( index_elements=[LinkEndPointModel.link_uuid, LinkEndPointModel.endpoint_uuid] ) session.execute(stmt) if len(related_topologies) > 0: session.execute(insert(TopologyLinkModel).values(related_topologies).on_conflict_do_nothing( index_elements=[TopologyLinkModel.topology_uuid, TopologyLinkModel.link_uuid] )) return updated updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_link_id(link_uuid),updated def link_delete(db_engine : Engine, request : LinkId) -> Tuple[Dict, bool]: link_uuid = link_get_uuid(request, allow_random=False) def callback(session : Session) -> bool: num_deleted = session.query(LinkModel).filter_by(link_uuid=link_uuid).delete() return num_deleted > 0 deleted = run_transaction(sessionmaker(bind=db_engine), callback) return json_link_id(link_uuid),deleted