# 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 import delete
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session
from typing import Dict, List, Optional
from common.proto.context_pb2 import ConfigRule
from common.tools.grpc.Tools import grpc_message_to_json_string
from .models.enums.ConfigAction import grpc_to_enum__config_action
from .models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel
from .uuids._Builder import get_uuid_random
def compose_config_rules_data(
config_rules : List[ConfigRule],
device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None
) -> List[Dict]:
dict_config_rules : List[Dict] = list()
for position,config_rule in enumerate(config_rules):
configrule_uuid = get_uuid_random()
str_kind = config_rule.WhichOneof('config_rule')
dict_config_rule = {
'configrule_uuid': configrule_uuid,
'position' : position,
'kind' : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member
'action' : grpc_to_enum__config_action(config_rule.action),
'data' : grpc_message_to_json_string(getattr(config_rule, str_kind, {})),
}
if device_uuid is not None: dict_config_rule['device_uuid' ] = device_uuid
if service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid
if slice_uuid is not None: dict_config_rule['slice_uuid' ] = slice_uuid
dict_config_rules.append(dict_config_rule)
return dict_config_rules
def upsert_config_rules(
session : Session, config_rules : List[Dict],
device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None
) -> None:
stmt = delete(ConfigRuleModel)
if device_uuid is not None: stmt = stmt.where(ConfigRuleModel.device_uuid == device_uuid )
if service_uuid is not None: stmt = stmt.where(ConfigRuleModel.service_uuid == service_uuid)
if slice_uuid is not None: stmt = stmt.where(ConfigRuleModel.slice_uuid == slice_uuid )
session.execute(stmt)
if len(config_rules) > 0:
session.execute(insert(ConfigRuleModel).values(config_rules))
#Union_SpecificConfigRule = Union[
# ConfigRuleCustomModel, ConfigRuleAclModel
#]
#
#def set_config_rule(
# database : Database, db_config : ConfigModel, position : int, resource_key : str, resource_value : str,
#): # -> Tuple[ConfigRuleModel, bool]:
#
# str_rule_key_hash = fast_hasher(resource_key)
# str_config_rule_key = key_to_str([db_config.config_uuid, str_rule_key_hash], separator=':')
#
# data = {'config_fk': db_config, 'position': position, 'action': ORM_ConfigActionEnum.SET, 'key': resource_key,
# 'value': resource_value}
# to_add = ConfigRuleModel(**data)
#
# result = database.create_or_update(to_add)
# return result
#Tuple_ConfigRuleSpecs = Tuple[Type, str, Dict, ConfigRuleKindEnum]
#
#def parse_config_rule_custom(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs:
# config_rule_class = ConfigRuleCustomModel
# str_config_rule_id = grpc_config_rule.custom.resource_key
# config_rule_data = {
# 'key' : grpc_config_rule.custom.resource_key,
# 'value': grpc_config_rule.custom.resource_value,
# }
# return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.CUSTOM
#
#def parse_config_rule_acl(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs:
# config_rule_class = ConfigRuleAclModel
# grpc_endpoint_id = grpc_config_rule.acl.endpoint_id
# grpc_rule_set = grpc_config_rule.acl.rule_set
# device_uuid = grpc_endpoint_id.device_id.device_uuid.uuid
# endpoint_uuid = grpc_endpoint_id.endpoint_uuid.uuid
# str_endpoint_key = '/'.join([device_uuid, endpoint_uuid])
# #str_endpoint_key, db_endpoint = get_endpoint(database, grpc_endpoint_id)
# str_config_rule_id = ':'.join([str_endpoint_key, grpc_rule_set.name])
# config_rule_data = {
# #'endpoint_fk': db_endpoint,
# 'endpoint_id': grpc_message_to_json_string(grpc_endpoint_id),
# 'acl_data': grpc_message_to_json_string(grpc_rule_set),
# }
# return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.ACL
#
#CONFIGRULE_PARSERS = {
# 'custom': parse_config_rule_custom,
# 'acl' : parse_config_rule_acl,
#}
#
#Union_ConfigRuleModel = Union[
# ConfigRuleCustomModel, ConfigRuleAclModel,
#]
#
#def set_config_rule(
# database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule, position : int
#) -> Tuple[Union_ConfigRuleModel, bool]:
# grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule'))
# parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind)
# if parser is None:
# raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format(
# grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule)))
#
# # create specific ConfigRule
# config_rule_class, str_config_rule_id, config_rule_data, config_rule_kind = parser(database, grpc_config_rule)
# str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id]))
# str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':')
# result : Tuple[Union_ConfigRuleModel, bool] = update_or_create_object(
# database, config_rule_class, str_config_rule_key, config_rule_data)
# db_specific_config_rule, updated = result
#
# # create generic ConfigRule
# config_rule_fk_field_name = 'config_rule_{:s}_fk'.format(config_rule_kind.value)
# config_rule_data = {
# 'config_fk': db_config, 'kind': config_rule_kind, 'position': position,
# 'action': ORM_ConfigActionEnum.SET,
# config_rule_fk_field_name: db_specific_config_rule
# }
# result : Tuple[ConfigRuleModel, bool] = update_or_create_object(
# database, ConfigRuleModel, str_config_rule_key, config_rule_data)
# db_config_rule, updated = result
#
# return db_config_rule, updated
#
#def delete_config_rule(
# database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule
#) -> None:
# grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule'))
# parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind)
# if parser is None:
# raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format(
# grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule)))
#
# # delete generic config rules; self deletes specific config rule
# _, str_config_rule_id, _, config_rule_kind = parser(database, grpc_config_rule)
# str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id]))
# str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':')
# db_config_rule : Optional[ConfigRuleModel] = get_object(
# database, ConfigRuleModel, str_config_rule_key, raise_if_not_found=False)
# if db_config_rule is None: return
# db_config_rule.delete()
#
#def update_config(
# database : Database, db_parent_pk : str, config_name : str, grpc_config_rules
#) -> List[Tuple[Union[ConfigModel, ConfigRuleModel], bool]]:
#
# str_config_key = key_to_str([config_name, db_parent_pk], separator=':')
# result : Tuple[ConfigModel, bool] = get_or_create_object(database, ConfigModel, str_config_key)
# db_config, created = result
#
# db_objects = [(db_config, created)]
#
# for position,grpc_config_rule in enumerate(grpc_config_rules):
# action = grpc_to_enum__config_action(grpc_config_rule.action)
#
# if action == ORM_ConfigActionEnum.SET:
# result : Tuple[ConfigRuleModel, bool] = set_config_rule(
# database, db_config, grpc_config_rule, position)
# db_config_rule, updated = result
# db_objects.append((db_config_rule, updated))
# elif action == ORM_ConfigActionEnum.DELETE:
# delete_config_rule(database, db_config, grpc_config_rule)
# else:
# msg = 'Unsupported Action({:s}) for ConfigRule({:s})'
# str_action = str(ConfigActionEnum.Name(action))
# str_config_rule = grpc_message_to_json_string(grpc_config_rule)
# raise AttributeError(msg.format(str_action, str_config_rule))
#
# return db_objects