# 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. from sqlalchemy.dialects.postgresql import insert from sqlalchemy.engine import Engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy_cockroachdb import run_transaction from typing import Dict, List, Optional, Set, Tuple from common.proto.context_pb2 import Link, LinkId, LinkIdList, LinkList 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 def link_list_ids(db_engine : Engine) -> LinkIdList: def callback(session : Session) -> List[Dict]: obj_list : List[LinkModel] = session.query(LinkModel).all() #.options(selectinload(LinkModel.topology)).filter_by(context_uuid=context_uuid).one_or_none() return [obj.dump_id() for obj in obj_list] return LinkIdList(link_ids=run_transaction(sessionmaker(bind=db_engine), callback)) def link_list_objs(db_engine : Engine) -> LinkList: def callback(session : Session) -> List[Dict]: obj_list : List[LinkModel] = session.query(LinkModel).all() #.options(selectinload(LinkModel.topology)).filter_by(context_uuid=context_uuid).one_or_none() return [obj.dump() for obj in obj_list] return LinkList(links=run_transaction(sessionmaker(bind=db_engine), callback)) def link_get(db_engine : Engine, request : LinkId) -> Link: link_uuid = link_get_uuid(request, allow_random=False) def callback(session : Session) -> Optional[Dict]: obj : Optional[LinkModel] = session.query(LinkModel).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 Link(**obj) def link_set(db_engine : Engine, request : Link) -> Tuple[LinkId, 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) topology_uuids : Set[str] = set() related_topologies : List[Dict] = list() link_endpoints_data : List[Dict] = list() for endpoint_id in 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, }) 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, }] def callback(session : Session) -> None: 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) ) session.execute(stmt) stmt = insert(LinkEndPointModel).values(link_endpoints_data) stmt = stmt.on_conflict_do_nothing( index_elements=[LinkEndPointModel.link_uuid, LinkEndPointModel.endpoint_uuid] ) session.execute(stmt) session.execute(insert(TopologyLinkModel).values(related_topologies).on_conflict_do_nothing( index_elements=[TopologyLinkModel.topology_uuid, TopologyLinkModel.link_uuid] )) run_transaction(sessionmaker(bind=db_engine), callback) updated = False # TODO: improve and check if created/updated return LinkId(**json_link_id(link_uuid)),updated def link_delete(db_engine : Engine, request : LinkId) -> 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 return run_transaction(sessionmaker(bind=db_engine), callback)