Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# 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)