# 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 uuid,json 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.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException from common.proto.context_pb2 import Device, DeviceFilter, DeviceId, TopologyId,MyConfig,MyConfigId,MyConfigList from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Device import json_device_id from context.service.database.uuids.Topology import topology_get_uuid from .models.DeviceModel import DeviceModel from .models.EndPointModel import EndPointModel from .models.TopologyModel import TopologyDeviceModel from .models.enums.DeviceDriver import grpc_to_enum__device_driver from .models.enums.DeviceOperationalStatus import grpc_to_enum__device_operational_status from .models.enums.KpiSampleType import grpc_to_enum__kpi_sample_type from .uuids.Device import device_get_uuid from .uuids.EndPoint import endpoint_get_uuid from .ConfigRule import compose_config_rules_data, upsert_config_rules from .models.MyConfigModel import MyConfigModel LOGGER = logging.getLogger(__name__) def get_myconfig (db_engine:Engine): def callback(session:Session): lst = list() results = session.query(MyConfigModel).all() for obj in results: myconfig=MyConfig() myconfig.config=json.dump(obj.config) myid=MyConfigId() myid.myconfig_uuid=obj.myconfig_uuid myconfig.myconfig_id.CopyFrom(myid) lst.append(myconfig) return lst obj=run_transaction(sessionmaker(bind=db_engine),callback) return obj def set_myconfig (db_engine:Engine,request:MyConfig): myconfig_id=MyConfigId() myconfig_id.myconfig_uuid=request.myconfig_id.myconfig_uuid my_config_data=[] if (request.config): channels=[] transceivers=[] config = json.loads(request.config) if ( "channels" in config and len(config["channels"])>0): channels=[channel['name']['index'] for channel in config["channels"]] if ("transceivers" in config and len(config["transceivers"]["transceiver"])>0): transceivers=[transceiver for transceiver in config["transceivers"]["transceiver"]] my_config_data=[ { "myconfig_uuid":request.myconfig_id.myconfig_uuid, "channels":channels, "transcievers":transceivers, "interfaces":json.dumps(config["interfaces"]["interface"]), "channel_namespace":config["channel_namespace"], "endpoints":[json.dumps(endpoint) for endpoint in config["endpoints"]], "frequency":config["frequency"] if "frequency" in config else 0, "operational_mode":config["operational_mode"] if "operational_mode" in config else 0, "output_power":config["output_power"] if "output_power" in config else '', } ] def callback(session:Session)->bool: stmt = insert(MyConfigModel).values(my_config_data) stmt = stmt.on_conflict_do_update( index_elements=[MyConfigModel.myconfig_uuid], set_=dict( channel_namespace=stmt.excluded.channel_namespace ) ) stmt = stmt.returning(MyConfigModel.myconfig_uuid) id = session.execute(stmt).fetchone() myconfig_id =run_transaction(sessionmaker(bind=db_engine),callback) return {'myconfig_uuid': myconfig_id} def select_MyConfig(db_engine:Engine,request:MyConfigId): def callback(session : Session) -> MyConfig: result=MyConfig() obj = session.query(MyConfigModel).filter_by(myconfig_uuid=request.myconfig_uuid).first() if (obj is not None): myid=MyConfigId() myid.myconfig_uuid=obj.myconfig_uuid result.config=json.dumps(obj.dump()) result.myconfig_id.CopyFrom(myid) return result return run_transaction(sessionmaker(bind=db_engine,expire_on_commit=False), callback)